661 lines
56 KiB
Python
661 lines
56 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import time
|
|
import uuid
|
|
from collections import Counter
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from agents.venture_discovery import RESEARCH_CATEGORIES, VentureDiscoveryService
|
|
from control_plane.events.bus import EventBus
|
|
from control_plane.ventures.models import (
|
|
AutonomousOperabilityAssessment,
|
|
AutonomousGateResult,
|
|
CompanyMandate,
|
|
CompanyProposal,
|
|
CompanyProposalStatus,
|
|
CryptoICDecisionType,
|
|
CryptoJurisdictionPolicy,
|
|
CryptoRedTeamAssessment,
|
|
CryptoScenarioLab,
|
|
FounderDependencyLevel,
|
|
OnchainNecessityAssessment,
|
|
PortfolioICReview,
|
|
PortfolioThesis,
|
|
PortfolioThesisStatus,
|
|
ProtocolThesis,
|
|
ProtocolValueCapture,
|
|
TokenDemandLoop,
|
|
TokenNecessityClassification,
|
|
TokenomicsSimulation,
|
|
TokenRedTeamFlag,
|
|
TokenUtilityAssessment,
|
|
VentureArtifact,
|
|
VentureCohort,
|
|
VentureCohortMember,
|
|
VentureGenerationRejection,
|
|
NoveltyGateDecision,
|
|
VentureThesis,
|
|
VentureTrack,
|
|
)
|
|
from model_router.policy import model_for_role
|
|
from model_router.providers import extract_json_object
|
|
from model_router.router import ModelCapability, ModelRequestContract, ModelRouter
|
|
|
|
|
|
CRYPTO_TERRITORIES = [
|
|
"DECENTRALIZED_AI_COMPUTE",
|
|
"AGENT_TO_AGENT_PAYMENTS",
|
|
"PROOF_ATTESTATION_MARKETS",
|
|
"DECENTRALIZED_DATA_MARKETS",
|
|
"SECURITY_STAKING_PROTOCOLS",
|
|
"MACHINE_REPUTATION",
|
|
"AUTONOMOUS_API_MARKETPLACES",
|
|
"DEPIN_COORDINATION",
|
|
"ONCHAIN_AGENT_COMMERCE",
|
|
"DECENTRALIZED_MODEL_SERVICES",
|
|
"ONCHAIN_CREDENTIALS",
|
|
"PROTOCOLIZED_ESCROW",
|
|
"CRYPTO_NATIVE_INTELLIGENCE",
|
|
"OPEN_CRYPTO_CATEGORY",
|
|
]
|
|
|
|
ALLOWED_TOKEN_UTILITIES = [
|
|
"protocol fee settlement",
|
|
"staking tied to measurable service quality",
|
|
"slashing / economic guarantees",
|
|
"access to scarce network resources",
|
|
"compute/data marketplace coordination",
|
|
"collateral",
|
|
"security budget",
|
|
"material protocol governance",
|
|
"contributor/provider rewards",
|
|
"reputation-backed economic participation",
|
|
"decentralized marketplace coordination",
|
|
"machine-to-machine payments",
|
|
"proof/attestation markets",
|
|
"protocol-owned infrastructure",
|
|
"incentive alignment for decentralized supply",
|
|
]
|
|
|
|
INSUFFICIENT_TOKEN_UTILITIES = ["community", "marketing", "speculative upside", "generic rewards", "token-gated access", "governance theater"]
|
|
|
|
CRYPTO_SCORE_DIMENSIONS = [
|
|
"TOKEN_NECESSITY",
|
|
"REAL_USAGE_DEMAND",
|
|
"ONCHAIN_NECESSITY",
|
|
"VALUE_ACCRUAL_QUALITY",
|
|
"NETWORK_EFFECT_POTENTIAL",
|
|
"TOKENOMICS_SUSTAINABILITY",
|
|
"BOOTSTRAPPABILITY",
|
|
"AUTONOMOUS_OPERABILITY",
|
|
"SECURITY_MODEL_QUALITY",
|
|
"REGULATORY_MANAGEABILITY",
|
|
]
|
|
|
|
CRYPTO_SCENARIOS = [
|
|
"oracle failure",
|
|
"validator/provider collusion",
|
|
"sybil attack",
|
|
"token price crash",
|
|
"liquidity collapse",
|
|
"spam attack",
|
|
"fee spike",
|
|
"slashing event",
|
|
"bad provider output",
|
|
"treasury exhaustion",
|
|
"emissions reduction",
|
|
"bridge dependency failure",
|
|
"smart contract exploit scenario",
|
|
"governance capture",
|
|
]
|
|
|
|
|
|
class CryptoVentureService:
|
|
def __init__(self, router: ModelRouter | None = None, bus: EventBus | None = None, *, web_research_available: bool = True, generation_model_hint: str | None = None, research_model_hint: str | None = None, final_ic_model_hint: str | None = None) -> None:
|
|
self.router = router
|
|
self.bus = bus or EventBus()
|
|
self.generation_model_hint = generation_model_hint or model_for_role("venture_ideation")
|
|
self.research_model_hint = research_model_hint or model_for_role("venture_research")
|
|
self.final_ic_model_hint = final_ic_model_hint or model_for_role("venture_portfolio_ic")
|
|
self.venture = VentureDiscoveryService(router, self.bus, web_research_available=web_research_available, research_model_hint=self.research_model_hint, ideation_model_hint=self.generation_model_hint)
|
|
|
|
def create_crypto_mandate(self) -> CompanyMandate:
|
|
mandate = CompanyMandate.objects.create(
|
|
objective="Generate crypto-native protocol ventures where onchain execution and a native token are genuinely necessary for real usage, not speculation.",
|
|
constraints={"venture_track": VentureTrack.CRYPTO_PROTOCOL, "no_token_sale": True, "no_fundraising": True, "no_mainnet_issuance": True, "no_us_targeted_activity": True, "testnet_or_local_only": True, "no_real_spend": True, "no_user_contact": True},
|
|
optimization_targets=["token necessity", "onchain necessity", "real usage demand", "value accrual", "sustainable tokenomics", "autonomous operability", "regulatory manageability", "security model quality"],
|
|
metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.1", "venture_track": VentureTrack.CRYPTO_PROTOCOL},
|
|
)
|
|
self._artifact(None, mandate, "CRYPTO_VENTURE_MANDATE", "Crypto Protocol Venture V0.1 Mandate", mandate.constraints, "Crypto protocol cohort mandate. No token sale, fundraising, or mainnet issuance.", "crypto_venture")
|
|
return mandate
|
|
|
|
def prepare_cohort(self, *, size: int, graph_run=None, concurrency: int = 2) -> VentureCohort:
|
|
mandate = self.create_crypto_mandate()
|
|
cohort_id = f"CPV01-{time.strftime('%Y%m%d%H%M%S')}-{uuid.uuid4().hex[:8]}"
|
|
return VentureCohort.objects.create(cohort_id=cohort_id, mandate=mandate, cohort_size=size, graph_run=graph_run, concurrency=concurrency, status="PREPARING_CRYPTO", graph_versions={"cohort": "crypto_venture_cohort v1"}, research_policy={"source_linked_evidence_required": True, "research_insufficient_if_no_sources": True}, scoring_policy={"dimensions": CRYPTO_SCORE_DIMENSIONS}, metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.1", "venture_track": VentureTrack.CRYPTO_PROTOCOL, "real_spend": 0, "real_customer_outreach": False, "token_sale": False, "fundraising": False, "mainnet_issuance": False})
|
|
|
|
def generate_protocols(self, cohort: VentureCohort) -> list[CompanyProposal]:
|
|
accepted = []
|
|
attempts = 0
|
|
for index in range(cohort.cohort_size):
|
|
territory = CRYPTO_TERRITORIES[index % len(CRYPTO_TERRITORIES)]
|
|
payload, source = self._protocol_payload(index, territory)
|
|
attempts += 1
|
|
proposal = self._create_proposal(cohort.mandate, payload, source, territory)
|
|
VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"territory": territory})
|
|
accepted.append(proposal)
|
|
cohort.metrics = {**cohort.metrics, "requested_protocol_count": cohort.cohort_size, "generation_attempts": attempts, "accepted_protocols": len(accepted), "duplicate_rejections": 0, "token_necessity_rejections": 0, "generation_sources": sorted({p.metadata.get("generation_source", "unknown") for p in accepted})}
|
|
cohort.status = "PROTOCOLS_GENERATED"
|
|
cohort.save(update_fields=["metrics", "status", "updated_at"])
|
|
return accepted
|
|
|
|
def novelty_gate(self, cohort: VentureCohort) -> dict[str, int]:
|
|
removed = 0
|
|
hard_excluded = 0
|
|
duplicate = 0
|
|
seen: dict[str, CompanyProposal] = {}
|
|
hard_terms = ["meme", "generic dex", "generic l1", "generic l2", "nft collection", "yield farm", "ponzi", "copycat launchpad", "speculative asset"]
|
|
for member in list(cohort.members.select_related("proposal").order_by("created_at")):
|
|
proposal = member.proposal
|
|
text = self._proposal_text(proposal)
|
|
fp = self._fingerprint(proposal.protocol_thesis.protocol_category + proposal.protocol_thesis.protocol_thesis + " ".join(proposal.protocol_thesis.utility_categories))
|
|
reason = ""
|
|
decision = None
|
|
if any(term in text for term in hard_terms):
|
|
hard_excluded += 1
|
|
reason = "hard-excluded crypto thesis"
|
|
decision = NoveltyGateDecision.REGENERATE_HARD_EXCLUSION
|
|
elif fp in seen:
|
|
duplicate += 1
|
|
reason = f"duplicate protocol/token utility thesis: {seen[fp].title}"
|
|
decision = NoveltyGateDecision.REGENERATE_DUPLICATE
|
|
if decision:
|
|
VentureGenerationRejection.objects.create(cohort=cohort, slot_index=member.rank or 0, attempt=1, decision=decision, reason=reason, candidate={"title": proposal.title, "protocol": proposal.protocol_thesis.protocol_thesis}, similarity_score=1.0)
|
|
member.delete()
|
|
proposal.status = CompanyProposalStatus.REJECTED
|
|
proposal.save(update_fields=["status", "updated_at"])
|
|
removed += 1
|
|
else:
|
|
seen[fp] = proposal
|
|
cohort.metrics = {**cohort.metrics, "crypto_novelty_removed": removed, "hard_exclusion_rejections": cohort.metrics.get("hard_exclusion_rejections", 0) + hard_excluded, "duplicate_rejections": cohort.metrics.get("duplicate_rejections", 0) + duplicate}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
return {"removed": removed, "hard_exclusion_rejections": hard_excluded, "duplicate_rejections": duplicate}
|
|
|
|
def regenerate_rejected_slots(self, cohort: VentureCohort, *, reason: str = "novelty_or_token_gate") -> list[CompanyProposal]:
|
|
created = []
|
|
attempts = 0
|
|
max_attempts = cohort.cohort_size * 3
|
|
while cohort.members.count() < cohort.cohort_size and attempts < max_attempts:
|
|
attempts += 1
|
|
index = cohort.members.count() + attempts
|
|
territory = CRYPTO_TERRITORIES[index % len(CRYPTO_TERRITORIES)]
|
|
payload, source = self._protocol_payload(index, territory)
|
|
proposal = self._create_proposal(cohort.mandate, payload, source, territory)
|
|
VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"territory": territory, "regenerated_for": reason})
|
|
created.append(proposal)
|
|
self.assess_token_necessity(proposal)
|
|
self.novelty_gate(cohort)
|
|
self._remove_weak_token_members(cohort)
|
|
cohort.metrics = {**cohort.metrics, "regeneration_attempts": cohort.metrics.get("regeneration_attempts", 0) + attempts, "regenerated_protocols": cohort.metrics.get("regenerated_protocols", 0) + len(created), "accepted_protocols": cohort.members.count()}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
return created
|
|
|
|
def token_necessity_gate(self, cohort: VentureCohort) -> list[TokenUtilityAssessment]:
|
|
assessments = [self.assess_token_necessity(member.proposal) for member in cohort.members.select_related("proposal")]
|
|
removed = self._remove_weak_token_members(cohort)
|
|
cohort.metrics = {**cohort.metrics, **removed}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
return assessments
|
|
|
|
def _remove_weak_token_members(self, cohort: VentureCohort) -> dict[str, int]:
|
|
unnecessary = 0
|
|
optional = 0
|
|
for member in list(cohort.members.select_related("proposal")):
|
|
proposal = member.proposal
|
|
assessment = getattr(proposal, "token_utility_assessment", None) or self.assess_token_necessity(proposal)
|
|
if assessment.classification not in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED}:
|
|
if assessment.classification == TokenNecessityClassification.TOKEN_OPTIONAL:
|
|
optional += 1
|
|
else:
|
|
unnecessary += 1
|
|
VentureGenerationRejection.objects.create(cohort=cohort, slot_index=member.rank or 0, attempt=1, decision=NoveltyGateDecision.REGENERATE_HARD_EXCLUSION, reason=f"removed by token necessity gate: {assessment.classification}", candidate={"title": proposal.title, "classification": assessment.classification}, similarity_score=0.0)
|
|
member.delete()
|
|
proposal.status = CompanyProposalStatus.REJECTED
|
|
proposal.metadata = {**proposal.metadata, "routed_to_saas": assessment.classification == TokenNecessityClassification.TOKEN_OPTIONAL}
|
|
proposal.save(update_fields=["status", "metadata", "updated_at"])
|
|
return {"token_unnecessary_rejections": cohort.metrics.get("token_unnecessary_rejections", 0) + unnecessary, "token_optional_route_to_saas": cohort.metrics.get("token_optional_route_to_saas", 0) + optional}
|
|
|
|
def light_market_research(self, cohort: VentureCohort) -> None:
|
|
started = time.monotonic()
|
|
total_sources = 0
|
|
insufficient = 0
|
|
for member in cohort.members.select_related("proposal"):
|
|
research = self.venture.conduct_market_research(member.proposal, depth="light")
|
|
total_sources += len(research.get("sources", []))
|
|
if not research.get("sources"):
|
|
insufficient += 1
|
|
cohort.metrics = {**cohort.metrics, "crypto_light_research_seconds": round(time.monotonic() - started, 2), "crypto_light_research_sources": total_sources, "research_insufficient_count": insufficient}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
|
|
def protocol_research(self, cohort: VentureCohort) -> None:
|
|
for member in cohort.members.select_related("proposal"):
|
|
self._update_protocol_research_flags(member.proposal)
|
|
|
|
def token_red_team(self, cohort: VentureCohort) -> list[CryptoRedTeamAssessment]:
|
|
assessments = [self.red_team_proposal(member.proposal) for member in cohort.members.select_related("proposal")]
|
|
counts = Counter(flag for item in assessments for flag in item.flags)
|
|
cohort.metrics = {**cohort.metrics, "token_red_team_failures": dict(counts)}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
return assessments
|
|
|
|
def tokenomics_simulation(self, cohort: VentureCohort) -> list[TokenomicsSimulation]:
|
|
simulations = [self.simulate_tokenomics(member.proposal) for member in cohort.members.select_related("proposal")]
|
|
self.crypto_scenario_lab(cohort)
|
|
return simulations
|
|
|
|
def crypto_ic_first_pass(self, cohort: VentureCohort) -> None:
|
|
for member in cohort.members.select_related("proposal"):
|
|
row = self.crypto_score_row(member.proposal)
|
|
member.portfolio_score = row["crypto_ic_score"]
|
|
member.metadata = {**member.metadata, "crypto_first_pass": row}
|
|
member.save(update_fields=["portfolio_score", "metadata", "updated_at"])
|
|
|
|
def top5_deep_research(self, cohort: VentureCohort) -> None:
|
|
before = {}
|
|
after = {}
|
|
members = list(cohort.members.select_related("proposal").order_by("-portfolio_score", "created_at")[:5])
|
|
for member in members:
|
|
before[str(member.proposal_id)] = member.proposal.metadata.get("research", {}).get("coverage_ratio", 0.0)
|
|
self.venture.conduct_market_research(member.proposal, depth="deep")
|
|
after[str(member.proposal_id)] = member.proposal.metadata.get("research", {}).get("coverage_ratio", 0.0)
|
|
cohort.metrics = {**cohort.metrics, "top5_deep_research_count": len(members), "top5_research_coverage_before": before, "top5_research_coverage_after": after}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
|
|
def portfolio_crypto_ic(self, cohort: VentureCohort) -> PortfolioICReview:
|
|
rows = [self.crypto_score_row(member.proposal) for member in cohort.members.select_related("proposal")]
|
|
rows = [self._sol_final_ic_adjustment(row) for row in rows]
|
|
rows.sort(key=lambda row: row["crypto_ic_score"], reverse=True)
|
|
top_3 = [row for row in rows if self._qualifies_finalist(row)][:3]
|
|
top_ids = {row["proposal_id"] for row in top_3}
|
|
for rank, row in enumerate(rows, start=1):
|
|
member = cohort.members.get(proposal_id=row["proposal_id"])
|
|
member.rank = rank
|
|
member.portfolio_score = row["crypto_ic_score"]
|
|
member.is_top_3 = row["proposal_id"] in top_ids
|
|
member.save(update_fields=["rank", "portfolio_score", "is_top_3", "updated_at"])
|
|
row["rank"] = rank
|
|
review, _ = PortfolioICReview.objects.update_or_create(cohort=cohort, defaults={"rankings": rows, "top_3": top_3, "concentration": self._crypto_concentration(rows), "metadata": {"finalist_shortfall": max(0, 3 - len(top_3)), "no_fund_decision_v01": True}})
|
|
cohort.metrics = {**cohort.metrics, "crypto_ranked_count": len(rows), "crypto_top_3_count": len(top_3)}
|
|
cohort.status = "CRYPTO_IC_COMPLETE"
|
|
cohort.save(update_fields=["metrics", "status", "updated_at"])
|
|
return review
|
|
|
|
def protocol_security_gate(self, cohort: VentureCohort) -> dict[str, int]:
|
|
blocked = 0
|
|
passed = 0
|
|
for member in cohort.members.select_related("proposal"):
|
|
proposal = member.proposal
|
|
text = self._proposal_text(proposal)
|
|
required = ["contract", "oracle", "key", "admin", "pause", "treasury"]
|
|
missing = [item for item in required if item not in text]
|
|
critical = any(self._positive_phrase_present(text, term) for term in ["unaudited mainnet", "custody user funds", "bridge dependency", "upgradeable without timelock"])
|
|
status = "BLOCKED_HUMAN_SECURITY_GATE" if critical or len(missing) >= 4 else "PASS_WITH_SECURITY_REVIEW"
|
|
if status.startswith("BLOCKED"):
|
|
blocked += 1
|
|
else:
|
|
passed += 1
|
|
self._artifact(proposal, cohort.mandate, "CRYPTO_PROTOCOL_SECURITY_GATE", f"Protocol Security Gate: {proposal.title}", {"status": status, "missing_controls": missing, "critical_risk": critical, "mainnet_allowed": False}, f"{status}. Missing controls: {', '.join(missing) or 'none'}. Mainnet is not allowed in V0.1.", "crypto_protocol_security_gate", graph_run=cohort.graph_run)
|
|
proposal.metadata = {**proposal.metadata, "crypto_security_gate": {"status": status, "missing_controls": missing, "critical_risk": critical, "mainnet_allowed": False}}
|
|
proposal.save(update_fields=["metadata", "updated_at"])
|
|
cohort.metrics = {**cohort.metrics, "protocol_security_gate_blocked": blocked, "protocol_security_gate_passed": passed}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
return {"blocked": blocked, "passed": passed}
|
|
|
|
def regulatory_gate(self, cohort: VentureCohort) -> None:
|
|
gated = 0
|
|
for member in cohort.members.select_related("proposal"):
|
|
policy = self.regulatory_policy(member.proposal)
|
|
if policy.human_legal_gate:
|
|
gated += 1
|
|
cohort.metrics = {**cohort.metrics, "human_legal_gate_count": gated, "legal_review_required": True}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
|
|
def capability_analysis(self, cohort: VentureCohort) -> list[dict[str, Any]]:
|
|
gaps = ["smart contract audit", "testnet deployment", "wallet auth", "key management", "oracle/provider monitoring", "token simulation harness", "legal review workflow"]
|
|
rows = [{"capability": gap, "count": cohort.members.count(), "status": "MISSING", "earliest_stage": "BEFORE_VALIDATION"} for gap in gaps]
|
|
review = getattr(cohort, "portfolio_review", None)
|
|
if review:
|
|
review.capability_demand = rows
|
|
review.recommended_build_priorities = rows[:5]
|
|
review.save(update_fields=["capability_demand", "recommended_build_priorities", "updated_at"])
|
|
return rows
|
|
|
|
def update_crypto_thesis_registry(self, cohort: VentureCohort) -> list[dict[str, Any]]:
|
|
updates = []
|
|
for member in cohort.members.select_related("proposal"):
|
|
proposal = member.proposal
|
|
proto = proposal.protocol_thesis
|
|
token = proposal.token_utility_assessment
|
|
name = f"CRYPTO::{proto.protocol_category or member.metadata.get('territory', 'OPEN')}::{proposal.title[:80]}"
|
|
registry, _ = PortfolioThesis.objects.update_or_create(canonical_name=name, defaults={"concise_thesis": proto.protocol_thesis[:1000], "category": proto.protocol_category, "industry": "crypto/protocol", "icp": proposal.target_customer[:1000], "problem": proposal.problem[:1000], "offer": proposal.proposed_solution[:1000], "business_model": "protocol", "primary_channel": "developer/community", "ai_leverage": 0.0, "platformization_potential": 0.0, "fingerprint": self._fingerprint(proposal.title + proto.protocol_thesis), "status": PortfolioThesisStatus.ACTIVE_CANDIDATE, "proposal_count": 1, "best_ic_score": member.portfolio_score, "best_company": proposal, "last_seen_cohort": cohort, "metadata": {"venture_track": VentureTrack.CRYPTO_PROTOCOL, "protocol_category": proto.protocol_category, "token_utility_type": token.utility_categories, "token_necessity": token.classification, "regulatory_risk": proposal.crypto_jurisdiction_policy.regulatory_manageability_score, "autonomy_score": proposal.autonomous_assessment.autonomous_operability_score}, "venture_track": VentureTrack.CRYPTO_PROTOCOL, "best_autonomous_operability_score": proposal.autonomous_assessment.autonomous_operability_score})
|
|
if registry.first_seen_cohort_id is None:
|
|
registry.first_seen_cohort = cohort
|
|
registry.save(update_fields=["first_seen_cohort", "updated_at"])
|
|
updates.append({"canonical_name": registry.canonical_name, "token_necessity": token.classification, "best_ic_score": member.portfolio_score})
|
|
cohort.metadata = {**cohort.metadata, "crypto_thesis_registry_after": updates}
|
|
cohort.save(update_fields=["metadata", "updated_at"])
|
|
return updates
|
|
|
|
def produce_crypto_cohort_report(self, cohort: VentureCohort) -> VentureArtifact:
|
|
review = cohort.portfolio_review
|
|
rows = review.rankings
|
|
content = {"title": "CRYPTO VENTURE COHORT V0.1 REPORT", "cohort_id": cohort.cohort_id, "graph_run": str(cohort.graph_run_id or ""), "runtime": cohort.metrics, "accepted_protocols": cohort.members.count(), "generation_attempts": cohort.metrics.get("generation_attempts", 0), "duplicate_token_necessity_rejections": {"duplicate_rejections": cohort.metrics.get("duplicate_rejections", 0), "token_unnecessary_rejections": cohort.metrics.get("token_unnecessary_rejections", 0)}, "ranking": rows, "top_3": review.top_3, "token_red_team_failures": cohort.metrics.get("token_red_team_failures", {}), "token_utility_distribution": dict(Counter(util for row in rows for util in row.get("token_utility", []))), "crypto_thesis_saturation": review.concentration, "capability_gaps": review.capability_demand, "system_metrics": cohort.metrics, "stop_conditions": {"token_sale": False, "fundraising": False, "mainnet_issuance": False, "user_contact": False, "real_spend": 0}}
|
|
artifact = self._artifact(None, cohort.mandate, "CRYPTO_VENTURE_COHORT_REPORT", "Crypto Venture Cohort V0.1 Report", content, self._readable_report(content), "crypto_portfolio_ic", graph_run=cohort.graph_run)
|
|
cohort.status = "COMPLETE"
|
|
cohort.save(update_fields=["status", "updated_at"])
|
|
return artifact
|
|
|
|
def assess_token_necessity(self, proposal: CompanyProposal) -> TokenUtilityAssessment:
|
|
crypto = proposal.metadata.get("crypto", {})
|
|
text = self._proposal_text(proposal)
|
|
utilities = [u for u in crypto.get("token_utility", []) if isinstance(u, str)] or self._infer_utilities(text)
|
|
strong_count = sum(1 for utility in utilities if utility.lower() in ALLOWED_TOKEN_UTILITIES or any(term in utility.lower() for term in ["stake", "slash", "collateral", "fee", "marketplace", "attestation", "compute", "machine", "security"]))
|
|
weak = any(term in text for term in ["meme", "speculative", "community token", "governance token only", "token gated subscription"])
|
|
score = min(100, 35 + strong_count * 18 + (10 if "slash" in text or "slashing" in text else 0) + (8 if "provider" in text else 0) - (35 if weak else 0))
|
|
sol_review = self._sol_json("Counterfactual token necessity review. If the token were removed and replaced with fiat/stablecoin/database credits, would the product materially degrade? Return JSON with classification TOKEN_ESSENTIAL, TOKEN_STRONGLY_JUSTIFIED, TOKEN_OPTIONAL, or TOKEN_UNNECESSARY; score 0-100; rationale; fiat_or_database_substitution. Proposal: " + json.dumps(proposal.pitch, default=str) + " Crypto: " + json.dumps(crypto, default=str))
|
|
if sol_review:
|
|
score = float(sol_review.get("score", score))
|
|
sol_classification = str(sol_review.get("classification", ""))
|
|
else:
|
|
sol_classification = ""
|
|
if score >= 82:
|
|
classification = TokenNecessityClassification.TOKEN_ESSENTIAL
|
|
elif score >= 70:
|
|
classification = TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED
|
|
elif score >= 50:
|
|
classification = TokenNecessityClassification.TOKEN_OPTIONAL
|
|
else:
|
|
classification = TokenNecessityClassification.TOKEN_UNNECESSARY
|
|
if sol_classification in TokenNecessityClassification.values:
|
|
classification = sol_classification
|
|
return TokenUtilityAssessment.objects.update_or_create(proposal=proposal, defaults={"classification": classification, "token_necessity_score": score, "utility_categories": utilities, "fiat_or_database_substitution": str((sol_review or {}).get("fiat_or_database_substitution", "If normal fiat/stablecoin/database credits preserve the core coordination, route to SaaS.")), "rationale": str((sol_review or {}).get("rationale", "Token score is based on explicit non-speculative utility, staking/slashing, marketplace coordination, and usage-linked fee demand.")), "metadata": {"weak_utility_terms": INSUFFICIENT_TOKEN_UTILITIES, "sol_counterfactual_review": sol_review or {}}})[0]
|
|
|
|
def _create_proposal(self, mandate: CompanyMandate, payload: dict[str, Any], source: str, territory: str) -> CompanyProposal:
|
|
title = str(payload.get("name") or payload.get("title") or f"Protocol {territory.title()}")[:255]
|
|
product = str(payload.get("product_thesis") or payload.get("product") or "A useful crypto-native network service.")
|
|
protocol = str(payload.get("protocol_thesis") or payload.get("protocol") or "Onchain settlement coordinates independent providers and users.")
|
|
token = str(payload.get("token_thesis") or payload.get("why_token") or "A native token bonds providers, pays protocol fees, and funds security.")
|
|
demand_loop = self._as_list(payload.get("token_demand_loop") or ["users consume service", "users pay protocol fees", "providers stake token", "bad providers are slashed", "usage-linked fees sustain rewards"])
|
|
validation = str(payload.get("validation_experiment") or "Run a testnet/local-chain pilot with fake credits and simulated token accounting; no sale, fundraising, or mainnet issuance.")
|
|
thesis = VentureThesis.objects.create(mandate=mandate, title=title, thesis=protocol, similarity_fingerprint=self._fingerprint(title + protocol), metadata={"source": source, "venture_track": VentureTrack.CRYPTO_PROTOCOL, "territory": territory}, evidence_tier="TIER_0_THESIS")
|
|
proposal = CompanyProposal.objects.create(mandate=mandate, thesis=thesis, title=title, description=product, problem=str(payload.get("user_pain") or payload.get("problem") or "Users lack trustworthy decentralized coordination."), target_customer=str(payload.get("user") or payload.get("target_user") or "Developers and network participants"), proposed_solution=str(payload.get("product") or product), business_model=str(payload.get("business_model") or "Protocol fees on real usage; no V0 token sale."), pricing_hypothesis=str(payload.get("fee_model") or "Testnet/free validation, later usage fees."), acquisition_strategy=str(payload.get("bootstrap_plan") or "Developer adoption through testnet docs and public artifacts only after approval."), validation_plan=validation, capital_requested=Decimal("0"), time_to_first_dollar_estimate="No V0 revenue target; validate protocol usage without token sale.", expected_margin="Protocol fee margin depends on provider economics.", build_complexity=str(payload.get("build_complexity") or "MEDIUM-HIGH"), market_evidence=[], differentiation=str(payload.get("differentiation") or protocol), major_risks=self._as_list(payload.get("major_risks") or ["Token not required", "Regulatory uncertainty", "Security model weak"]), confidence=float(payload.get("confidence") or 0.55), status=CompanyProposalStatus.SUBMITTED, pitch={"Company name": title, "One-line thesis": str(payload.get("one_line_thesis") or protocol), "Product thesis": product, "Protocol thesis": protocol, "Token thesis": token, "Token demand loop": demand_loop, "Validation experiment": validation}, metadata={"generation_source": source, "venture_track": VentureTrack.CRYPTO_PROTOCOL, "crypto": {"territory": territory, "product_thesis": product, "protocol_thesis": protocol, "token_thesis": token, "token_demand_loop": demand_loop, "token_utility": self._as_list(payload.get("token_utility") or self._infer_utilities(token + " " + protocol)), "value_capture": str(payload.get("value_capture") or "Usage fees accrue to providers, security budget, and protocol treasury."), "network_effect": str(payload.get("network_effect") or "More users attract more providers, improving liquidity/reliability."), "bootstrap_plan": str(payload.get("bootstrap_plan") or "Run without a live token using test credits and provider simulations."), "security_model": str(payload.get("security_model") or "Contracts, staking, slashing, oracle controls, admin keys, and treasury controls require review."), "regulatory_policy": "US_EXCLUDED; TOKEN_SALE_DISABLED; MAINNET_TOKEN_ISSUANCE_DISABLED; FUNDRAISING_DISABLED"}, "real_spend": 0, "real_customer_outreach": False, "token_sale": False, "mainnet_issuance": False}, evidence_tier="TIER_0_THESIS")
|
|
ProtocolThesis.objects.create(proposal=proposal, product_thesis=product, protocol_thesis=protocol, token_thesis=token, network_effect=proposal.metadata["crypto"]["network_effect"], bootstrap_plan=proposal.metadata["crypto"]["bootstrap_plan"], autonomous_operability="Artifex can build/testnet deploy/monitor only; mainnet and issuance stop at human legal gate.", utility_categories=proposal.metadata["crypto"]["token_utility"], protocol_category=territory)
|
|
TokenDemandLoop.objects.create(proposal=proposal, loop=demand_loop, real_usage_driver=str(payload.get("real_usage_driver") or product), non_speculative_demand=not any(term in " ".join(demand_loop).lower() for term in ["speculation", "price go up"]), bootstrap_without_token=proposal.metadata["crypto"]["bootstrap_plan"])
|
|
self.bus.publish("CRYPTO_PROTOCOL_PROPOSED", payload={"proposal_id": str(proposal.id), "territory": territory, "source": source})
|
|
return proposal
|
|
|
|
def _protocol_payload(self, index: int, territory: str) -> tuple[dict[str, Any], str]:
|
|
if self.router is not None:
|
|
try:
|
|
prompt = "Generate exactly one crypto-native protocol venture as JSON. Do not propose meme coins, generic DEX/L1/NFT/yield farms, or SaaS plus token. Include name, one_line_thesis, product_thesis, user, protocol_thesis, why_onchain, token_thesis, token_utility list, token_demand_loop list, value_capture, network_effect, bootstrap_plan, validation_experiment, security_model, regulatory_risks, major_risks, confidence. Territory: " + territory + ". Rules: no token sale, no fundraising, no mainnet issuance, no US-targeted activity; validation must use testnet/local/fake credits/stablecoin-only simulation."
|
|
response = self.router.complete(ModelRequestContract(purpose=ModelCapability.PLANNING, model_hint=self.generation_model_hint, prompt=prompt))
|
|
parsed = extract_json_object(response.content)
|
|
if parsed:
|
|
return parsed, self.generation_model_hint
|
|
except Exception:
|
|
pass
|
|
return self._fallback_payload(index, territory), "deterministic_crypto_fallback"
|
|
|
|
def _fallback_payload(self, index: int, territory: str) -> dict[str, Any]:
|
|
names = {
|
|
"DECENTRALIZED_AI_COMPUTE": "Verifiable Inference Provider Market",
|
|
"AGENT_TO_AGENT_PAYMENTS": "Agent Micropayment Settlement Rail",
|
|
"PROOF_ATTESTATION_MARKETS": "Model Output Attestation Market",
|
|
"DECENTRALIZED_DATA_MARKETS": "Consent-Bound Data License Exchange",
|
|
"SECURITY_STAKING_PROTOCOLS": "API Reliability Slashing Pool",
|
|
}
|
|
name = names.get(territory, f"{territory.replace('_', ' ').title()} Protocol")
|
|
return {"name": name, "one_line_thesis": f"{name} coordinates independent supply and demand with staking, slashing, and usage fees.", "product_thesis": "A testnet marketplace/API where users request measurable digital work and providers compete to fulfill it.", "user": "Developers, agents, and protocol operators needing verifiable digital services.", "protocol_thesis": "Onchain settlement, escrow, provider bonds, attestations, and slashing coordinate parties that do not share an operator.", "why_onchain": "Trust-minimized escrow, programmable slashing, public reputation, and machine-to-machine settlement materially degrade if replaced by a private database.", "token_thesis": "The token is staked by providers, slashed for measurable failures, used for protocol fee settlement, and funds the security budget.", "token_utility": ["protocol fee settlement", "staking tied to measurable service quality", "slashing / economic guarantees", "decentralized marketplace coordination"], "token_demand_loop": ["users request measurable service", "users pay protocol fee", "providers stake token to serve", "bad providers are slashed", "fees reward reliable providers and security budget", "more real usage increases fee demand"], "value_capture": "Usage fees and slashing penalties accrue to reliable providers, insurance/security pool, and protocol treasury.", "network_effect": "More users create more jobs; more staked providers improve reliability and lower latency; more attestations improve reputation quality.", "bootstrap_plan": "Validate on local/testnet with fake credits and recruited simulated providers; no token sale or mainnet issuance.", "validation_experiment": "Run 50 simulated jobs on testnet/local chain, measure provider quality, slashing events, completion cost, and developer API reuse.", "security_model": "Escrow contracts, staking/slashing, oracle/attestation checks, admin key limits, pause controls, and treasury multisig require review.", "major_risks": ["Token may be optional", "Provider supply bootstrapping", "Smart contract risk", "Regulatory uncertainty"], "confidence": 0.58 + (index % 3) * 0.04}
|
|
|
|
def _update_protocol_research_flags(self, proposal: CompanyProposal) -> None:
|
|
research = proposal.metadata.get("research", {})
|
|
crypto = proposal.metadata.get("crypto", {})
|
|
proposal.metadata = {**proposal.metadata, "crypto": {**crypto, "research_status": "RESEARCH_INSUFFICIENT" if research.get("source_count", 0) == 0 else "SOURCE_LINKED", "research_categories": RESEARCH_CATEGORIES}}
|
|
proposal.save(update_fields=["metadata", "updated_at"])
|
|
|
|
def red_team_proposal(self, proposal: CompanyProposal) -> CryptoRedTeamAssessment:
|
|
token = proposal.token_utility_assessment
|
|
onchain = self.onchain_assessment(proposal)
|
|
value = self.value_capture_assessment(proposal)
|
|
policy = self.regulatory_policy(proposal)
|
|
flags = []
|
|
if token.classification in {TokenNecessityClassification.TOKEN_OPTIONAL, TokenNecessityClassification.TOKEN_UNNECESSARY}:
|
|
flags.append(TokenRedTeamFlag.TOKEN_NOT_REQUIRED)
|
|
if onchain.onchain_necessity_score < 70:
|
|
flags.append(TokenRedTeamFlag.ONCHAIN_NOT_REQUIRED)
|
|
if value.value_accrual_quality_score < 65:
|
|
flags.append(TokenRedTeamFlag.VALUE_CAPTURE_BROKEN)
|
|
if value.tokenomics_sustainability_score < 65:
|
|
flags.append(TokenRedTeamFlag.UNSUSTAINABLE_EMISSIONS)
|
|
if policy.regulatory_manageability_score < 55:
|
|
flags.append(TokenRedTeamFlag.REGULATORY_RISK_HIGH)
|
|
text = self._proposal_text(proposal)
|
|
if "speculat" in text or "apy" in text or "yield farm" in text:
|
|
flags.append(TokenRedTeamFlag.SPECULATION_DEPENDENT)
|
|
if "governance" in text and "slash" not in text and "fee" not in text:
|
|
flags.append(TokenRedTeamFlag.GOVERNANCE_THEATER)
|
|
sol_review = self._sol_json("Independent Token Red Team. Return JSON with flags list using only TOKEN_NOT_REQUIRED, SPECULATION_DEPENDENT, UNSUSTAINABLE_EMISSIONS, VALUE_CAPTURE_BROKEN, MERCENARY_INCENTIVES, GOVERNANCE_THEATER, SECURITY_MODEL_WEAK, TOKEN_VELOCITY_TOO_HIGH, BOOTSTRAP_PROBLEM, CENTRALIZATION_CONTRADICTION, REGULATORY_RISK_HIGH, ONCHAIN_NOT_REQUIRED, NO_REAL_USER_DEMAND; severity LOW/MEDIUM/HIGH; critique. Proposal: " + json.dumps(self.crypto_score_context(proposal), default=str))
|
|
if sol_review:
|
|
for flag in self._as_list(sol_review.get("flags")):
|
|
if str(flag) in TokenRedTeamFlag.values:
|
|
flags.append(str(flag))
|
|
severity = "HIGH" if len(flags) >= 3 else "MEDIUM" if flags else "LOW"
|
|
if sol_review and str(sol_review.get("severity", "")) in {"LOW", "MEDIUM", "HIGH"}:
|
|
severity = str(sol_review["severity"])
|
|
deduped_flags = list(dict.fromkeys(str(flag) for flag in flags))
|
|
return CryptoRedTeamAssessment.objects.update_or_create(proposal=proposal, defaults={"flags": deduped_flags, "severity": severity, "critique": str((sol_review or {}).get("critique", "Independent token red team checked token necessity, speculation dependence, value capture, emissions, governance theater, onchain necessity, and regulatory risk.")), "independent_from_generation": True, "metadata": {"sol_red_team_review": sol_review or {}}})[0]
|
|
|
|
def crypto_score_context(self, proposal: CompanyProposal) -> dict[str, Any]:
|
|
return {"title": proposal.title, "pitch": proposal.pitch, "crypto": proposal.metadata.get("crypto", {}), "token": getattr(proposal, "token_utility_assessment", None).classification if hasattr(proposal, "token_utility_assessment") else None}
|
|
|
|
def onchain_assessment(self, proposal: CompanyProposal) -> OnchainNecessityAssessment:
|
|
text = self._proposal_text(proposal)
|
|
terms = ["escrow", "slashing", "stake", "attestation", "settlement", "collateral", "trust", "reputation", "marketplace", "machine"]
|
|
score = min(100, 35 + sum(8 for term in terms if term in text))
|
|
return OnchainNecessityAssessment.objects.update_or_create(proposal=proposal, defaults={"onchain_necessity_score": score, "reasons": [term for term in terms if term in text], "offchain_substitute": "Private SaaS/database is acceptable only if escrow, slashing, public reputation, and neutral settlement are not material.", "rationale": "Scores onchain necessity from trust-minimized coordination, settlement, staking/slashing, attestations, and multi-party neutrality."})[0]
|
|
|
|
def value_capture_assessment(self, proposal: CompanyProposal) -> ProtocolValueCapture:
|
|
text = self._proposal_text(proposal)
|
|
fee = "fee" in text
|
|
stake = "stake" in text or "staking" in text
|
|
slash = "slash" in text or "slashing" in text
|
|
emission_bad = any(term in text for term in ["high apy", "ponzi", "yield farm", "emissions only"])
|
|
value_score = min(100, 40 + (18 if fee else 0) + (14 if stake else 0) + (14 if slash else 0) + (10 if "treasury" in text or "security budget" in text else 0))
|
|
sustainability = max(0, min(100, value_score - (35 if emission_bad else 0) + (8 if "usage" in text else 0)))
|
|
return ProtocolValueCapture.objects.update_or_create(proposal=proposal, defaults={"value_accrual_quality_score": value_score, "tokenomics_sustainability_score": sustainability, "value_accrual": proposal.metadata.get("crypto", {}).get("value_capture", "Usage-linked fees are required."), "sinks": ["protocol fees", "staking bonds", "slashing penalties", "security budget"], "emissions_policy": "No high APY. Rewards must be covered by real fees after subsidies decline.", "sustainability_rationale": "High emissions without real fee demand score poorly; usage-linked fees, staking, slashing, and sinks score better."})[0]
|
|
|
|
def regulatory_policy(self, proposal: CompanyProposal) -> CryptoJurisdictionPolicy:
|
|
text = self._proposal_text(proposal)
|
|
issuance_risk = any(self._positive_phrase_present(text, term) for term in ["token sale", "public sale", "airdrop", "mainnet issuance", "fundraise", "fundraising"])
|
|
score = 55 if issuance_risk else 72
|
|
return CryptoJurisdictionPolicy.objects.update_or_create(proposal=proposal, defaults={"excluded_jurisdictions": ["US"], "excluded_person_classes": ["US persons", "sanctioned persons", "restricted jurisdictions"], "marketing_restrictions": ["No U.S.-targeted activity", "No investment marketing", "No yield/APY promises"], "sale_restrictions": ["TOKEN_SALE_DISABLED", "MAINNET_TOKEN_ISSUANCE_DISABLED", "FUNDRAISING_DISABLED"], "kyc_aml_requirement_status": "REQUIRES_LEGAL_REVIEW_BEFORE_ANY_TRANSFER_OR_SALE", "transfer_restriction_requirement_status": "UNDETERMINED_REQUIRES_COUNSEL", "legal_review_required": True, "regulatory_manageability_score": score, "jurisdiction_uncertainty": ["Blocking the USA does not remove all legal obligations", "Token transfer and marketing treatment requires counsel"], "human_legal_gate": True, "metadata": {"US_EXCLUDED": True, "TOKEN_SALE_DISABLED": True, "MAINNET_TOKEN_ISSUANCE_DISABLED": True, "FUNDRAISING_DISABLED": True}})[0]
|
|
|
|
def simulate_tokenomics(self, proposal: CompanyProposal) -> TokenomicsSimulation:
|
|
scenarios = {}
|
|
text = self._proposal_text(proposal)
|
|
token = getattr(proposal, "token_utility_assessment", None) or self.assess_token_necessity(proposal)
|
|
value = getattr(proposal, "value_capture", None) or self.value_capture_assessment(proposal)
|
|
base_users = max(20, int(float(proposal.confidence) * 180))
|
|
fee_per_tx = 0.04 + (token.token_necessity_score / 1000) + (0.03 if "enterprise" in text or "infrastructure" in text else 0)
|
|
tx_per_user = 6 + len(token.utility_categories) * 2 + (4 if "agent" in text or "machine" in text else 0)
|
|
staking_per_provider = 150 + int(value.value_accrual_quality_score * 8)
|
|
emission_rate = max(0.005, (100 - value.tokenomics_sustainability_score) / 1400)
|
|
for name, multiplier in {"LOW_USAGE": 0.3, "EXPECTED": 1.0, "HIGH_USAGE": 3.0, "SUBSIDY_REMOVED": 0.8, "TOKEN_PRICE_DOWN_80_PERCENT": 0.8, "PROVIDER_CHURN": 0.6, "USER_GROWTH_10X": 10.0}.items():
|
|
users = int(base_users * multiplier)
|
|
tx = int(users * tx_per_user)
|
|
fees = round(tx * fee_per_tx, 2)
|
|
provider_supply = max(3, int(users / (15 if "market" in text else 25)))
|
|
staking = provider_supply * staking_per_provider
|
|
emissions = 0 if name == "SUBSIDY_REMOVED" else round(max(3, tx * emission_rate), 2)
|
|
if name == "PROVIDER_CHURN":
|
|
provider_supply = max(1, int(provider_supply * 0.45))
|
|
staking = max(100, int(staking * 0.45))
|
|
scenarios[name] = {"users": users, "transactions": tx, "fees": fees, "token_demand": fees + staking * 0.01, "provider_supply": provider_supply, "staking": staking, "emissions": emissions, "treasury": round(1000 + fees * 0.2 - emissions * 0.1, 2), "circulating_supply": 1_000_000 + emissions, "token_velocity": round(tx / max(1, fees + staking * 0.01), 2), "reward_coverage": round(fees / max(1, emissions), 2), "network_security_budget": staking}
|
|
summary = {"token_price_appreciation_primary_success_variable": False, "subsidy_removed_survives": scenarios["SUBSIDY_REMOVED"]["reward_coverage"] >= 1.0, "stress_notes": ["Price-down and provider-churn scenarios must preserve service quality before mainnet."]}
|
|
return TokenomicsSimulation.objects.update_or_create(proposal=proposal, defaults={"scenarios": scenarios, "summary": summary})[0]
|
|
|
|
def crypto_scenario_lab(self, cohort: VentureCohort) -> list[CryptoScenarioLab]:
|
|
labs = []
|
|
for member in cohort.members.select_related("proposal"):
|
|
scenarios = [{"scenario": name, "expected_failure_mode": "Must be mitigated before any mainnet deployment.", "route_to_progeny": name in {"smart contract exploit scenario", "oracle failure", "governance capture"}} for name in CRYPTO_SCENARIOS]
|
|
labs.append(CryptoScenarioLab.objects.update_or_create(proposal=member.proposal, defaults={"scenarios": scenarios, "systemic_findings": ["No V0.1 mainnet deployment", "Audit/key-management/legal gates required"], "progeny_candidates": [s["scenario"] for s in scenarios if s["route_to_progeny"]]})[0])
|
|
return labs
|
|
|
|
def crypto_score_row(self, proposal: CompanyProposal) -> dict[str, Any]:
|
|
token = self.assess_token_necessity(proposal)
|
|
onchain = self.onchain_assessment(proposal)
|
|
value = self.value_capture_assessment(proposal)
|
|
policy = self.regulatory_policy(proposal)
|
|
red = getattr(proposal, "crypto_red_team", None) or self.red_team_proposal(proposal)
|
|
sim = getattr(proposal, "tokenomics_simulation", None) or self.simulate_tokenomics(proposal)
|
|
autonomy = self.assess_autonomy(proposal)
|
|
text = self._proposal_text(proposal)
|
|
component_scores = {
|
|
"TOKEN_NECESSITY": token.token_necessity_score,
|
|
"REAL_USAGE_DEMAND": min(100, 45 + (15 if "usage" in text else 0) + (10 if "users" in text else 0) + (10 if proposal.metadata.get("research", {}).get("source_count", 0) else 0)),
|
|
"ONCHAIN_NECESSITY": onchain.onchain_necessity_score,
|
|
"VALUE_ACCRUAL_QUALITY": value.value_accrual_quality_score,
|
|
"NETWORK_EFFECT_POTENTIAL": min(100, 50 + (20 if "provider" in text else 0) + (15 if "reputation" in text else 0)),
|
|
"TOKENOMICS_SUSTAINABILITY": value.tokenomics_sustainability_score,
|
|
"BOOTSTRAPPABILITY": 78 if "testnet" in text or "fake credits" in text else 55,
|
|
"AUTONOMOUS_OPERABILITY": autonomy.autonomous_operability_score,
|
|
"SECURITY_MODEL_QUALITY": min(100, 45 + sum(8 for term in ["contract", "oracle", "key", "pause", "treasury", "slashing"] if term in text)),
|
|
"REGULATORY_MANAGEABILITY": policy.regulatory_manageability_score,
|
|
}
|
|
score = round(sum(component_scores.values()) / len(component_scores) - len(red.flags) * 4, 1)
|
|
decision = self._crypto_decision(component_scores, token.classification, red.flags)
|
|
return {"proposal_id": str(proposal.id), "company": proposal.title, "one_line_thesis": proposal.pitch.get("One-line thesis", ""), "product_thesis": proposal.protocol_thesis.product_thesis, "protocol_thesis": proposal.protocol_thesis.protocol_thesis, "token_thesis": proposal.protocol_thesis.token_thesis, "why_onchain": proposal.onchain_assessment.rationale, "why_token": token.rationale, "token_utility": token.utility_categories, "token_necessity_classification": token.classification, "token_demand_loop": proposal.token_demand_loop.loop, "value_capture": proposal.value_capture.value_accrual, "network_effect": proposal.protocol_thesis.network_effect, "bootstrap_plan": proposal.protocol_thesis.bootstrap_plan, "autonomous_operability": autonomy.autonomous_operability_score, "regulatory_manageability": policy.regulatory_manageability_score, "security_risk": component_scores["SECURITY_MODEL_QUALITY"], "validation_experiment": proposal.validation_plan, "component_scores": component_scores, "value_accrual_quality": value.value_accrual_quality_score, "tokenomics_sustainability": value.tokenomics_sustainability_score, "simulation_summary": sim.summary, "token_red_team_flags": red.flags, "crypto_ic_decision": decision, "crypto_ic_score": score, "research": proposal.metadata.get("research", {}), "legal_review_required": policy.legal_review_required}
|
|
|
|
def assess_autonomy(self, proposal: CompanyProposal) -> AutonomousOperabilityAssessment:
|
|
assessment = self.venture.assess_autonomous_operability(proposal)
|
|
text = self._proposal_text(proposal)
|
|
penalty = 25 if any(term in text for term in ["exchange listing", "market making", "institutional integration", "founder evangelism"]) else 0
|
|
score = max(0, assessment.autonomous_operability_score - penalty)
|
|
minutes = assessment.minutes_per_week_human + (30 if penalty else 0)
|
|
gate = AutonomousGateResult.AUTONOMOUS_ELIGIBLE if score >= 75 and minutes <= 30 else AutonomousGateResult.AUTONOMOUS_BORDERLINE if score >= 60 else AutonomousGateResult.ASSISTED_ONLY
|
|
assessment.venture_track = VentureTrack.CRYPTO_PROTOCOL
|
|
assessment.gate_result = gate
|
|
assessment.autonomous_operability_score = score
|
|
assessment.minutes_per_week_human = minutes
|
|
assessment.human_actions_required = list(dict.fromkeys([*assessment.human_actions_required, "legal review", "security review", "mainnet/issuance approval gate"]))
|
|
assessment.human_action_categories = list(dict.fromkeys([*assessment.human_action_categories, "LEGAL_GATE", "SECURITY_GATE"]))
|
|
assessment.platform_blockers = list(dict.fromkeys([*assessment.platform_blockers, "LEGAL_REVIEW_REQUIRED", "SMART_CONTRACT_AUDIT", "KEY_MANAGEMENT", "TESTNET_DEPLOYMENT"]))
|
|
assessment.validation_offer = {**assessment.validation_offer, "type": "testnet/local simulation", "token_required": False}
|
|
assessment.end_state_business_model = {**assessment.end_state_business_model, "type": "usage-fee protocol", "token_sale": False, "mainnet_requires_human_gate": True}
|
|
assessment.rationale = assessment.rationale + " Crypto overlay: V0.1 can build/testnet/simulate autonomously, but issuance, fundraising, and mainnet deployment stop at legal/security gates."
|
|
assessment.save(update_fields=["venture_track", "gate_result", "autonomous_operability_score", "minutes_per_week_human", "human_actions_required", "human_action_categories", "platform_blockers", "validation_offer", "end_state_business_model", "rationale", "updated_at"])
|
|
return assessment
|
|
|
|
def _crypto_decision(self, scores: dict[str, float], classification: str, flags: list[str]) -> str:
|
|
if classification == TokenNecessityClassification.TOKEN_UNNECESSARY:
|
|
return CryptoICDecisionType.REJECT_TOKEN_NOT_NEEDED
|
|
if classification == TokenNecessityClassification.TOKEN_OPTIONAL:
|
|
return CryptoICDecisionType.ROUTE_TO_SAAS
|
|
if TokenRedTeamFlag.SPECULATION_DEPENDENT in flags:
|
|
return CryptoICDecisionType.REJECT_SPECULATIVE
|
|
if scores["REGULATORY_MANAGEABILITY"] < 55:
|
|
return CryptoICDecisionType.REJECT_REGULATORY_RISK
|
|
if scores["TOKENOMICS_SUSTAINABILITY"] < 65 or scores["VALUE_ACCRUAL_QUALITY"] < 65:
|
|
return CryptoICDecisionType.REVISE_TOKEN_MODEL
|
|
if scores["TOKEN_NECESSITY"] >= 82 and scores["ONCHAIN_NECESSITY"] >= 75:
|
|
return CryptoICDecisionType.TESTNET_PILOT
|
|
return CryptoICDecisionType.PROTOCOL_VALIDATE
|
|
|
|
def _sol_final_ic_adjustment(self, row: dict[str, Any]) -> dict[str, Any]:
|
|
review = self._sol_json("Final Crypto IC review. Return JSON with optional score_adjustment -20..20, decision from PROTOCOL_VALIDATE, TESTNET_PILOT, REVISE_TOKEN_MODEL, ROUTE_TO_SAAS, WATCHLIST, REJECT_TOKEN_NOT_NEEDED, REJECT_SPECULATIVE, REJECT_REGULATORY_RISK, REJECT_ECONOMIC_MODEL, REJECT_OPERABILITY, and rationale. Do not promote weak token ideas. Row: " + json.dumps(row, default=str))
|
|
if not review:
|
|
return row
|
|
adjustment = max(-20, min(20, float(review.get("score_adjustment", 0))))
|
|
decision = str(review.get("decision", row["crypto_ic_decision"]))
|
|
if decision not in CryptoICDecisionType.values:
|
|
decision = row["crypto_ic_decision"]
|
|
return {**row, "crypto_ic_score": round(max(0, min(100, row["crypto_ic_score"] + adjustment)), 1), "crypto_ic_decision": decision, "sol_final_ic_review": review}
|
|
|
|
def _qualifies_finalist(self, row: dict[str, Any]) -> bool:
|
|
scores = row["component_scores"]
|
|
return row["token_necessity_classification"] in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED} and scores["TOKEN_NECESSITY"] >= 75 and scores["REAL_USAGE_DEMAND"] >= 65 and scores["ONCHAIN_NECESSITY"] >= 70 and scores["VALUE_ACCRUAL_QUALITY"] >= 65 and scores["TOKENOMICS_SUSTAINABILITY"] >= 65 and scores["AUTONOMOUS_OPERABILITY"] >= 70
|
|
|
|
def _crypto_concentration(self, rows: list[dict[str, Any]]) -> dict[str, Any]:
|
|
categories = Counter(row["company"].split()[0] for row in rows)
|
|
utilities = Counter(util for row in rows for util in row.get("token_utility", []))
|
|
return {"protocol_category_distribution": dict(categories), "token_utility_distribution": dict(utilities), "saturation_flags": [name for name, count in categories.items() if count >= 3]}
|
|
|
|
def _infer_utilities(self, text: str) -> list[str]:
|
|
lowered = text.lower()
|
|
utilities = []
|
|
if "fee" in lowered or "settlement" in lowered:
|
|
utilities.append("protocol fee settlement")
|
|
if "stake" in lowered:
|
|
utilities.append("staking tied to measurable service quality")
|
|
if "slash" in lowered:
|
|
utilities.append("slashing / economic guarantees")
|
|
if "market" in lowered or "provider" in lowered:
|
|
utilities.append("decentralized marketplace coordination")
|
|
if "attestation" in lowered or "proof" in lowered:
|
|
utilities.append("proof/attestation markets")
|
|
return utilities or ["token utility not proven"]
|
|
|
|
def _proposal_text(self, proposal: CompanyProposal) -> str:
|
|
crypto = proposal.metadata.get("crypto", {}) if isinstance(proposal.metadata, dict) else {}
|
|
return " ".join([proposal.title, proposal.description, proposal.problem, proposal.target_customer, proposal.proposed_solution, proposal.business_model, proposal.validation_plan, json.dumps(crypto, default=str)]).lower()
|
|
|
|
def _positive_phrase_present(self, text: str, phrase: str) -> bool:
|
|
phrase = phrase.lower()
|
|
for match in re.finditer(re.escape(phrase), text):
|
|
prefix = text[max(0, match.start() - 40) : match.start()]
|
|
if any(negation in prefix for negation in ["no ", "not ", "without ", "disabled", "disable", "prohibit", "forbid", "never "]):
|
|
continue
|
|
return True
|
|
return False
|
|
|
|
def _sol_json(self, prompt: str) -> dict[str, Any]:
|
|
if self.router is None or self.final_ic_model_hint not in self.router.providers:
|
|
return {}
|
|
try:
|
|
response = self.router.complete(ModelRequestContract(purpose=ModelCapability.REASONING, model_hint=self.final_ic_model_hint, prompt=prompt))
|
|
parsed = extract_json_object(response.content)
|
|
return parsed if isinstance(parsed, dict) else {}
|
|
except Exception:
|
|
return {}
|
|
|
|
def _as_list(self, value: Any) -> list[Any]:
|
|
if isinstance(value, list):
|
|
return value
|
|
if value in (None, ""):
|
|
return []
|
|
return [value]
|
|
|
|
def _fingerprint(self, value: str) -> str:
|
|
tokens = sorted(set(re.findall(r"[a-z0-9]{4,}", value.lower())))
|
|
return hashlib.sha256("|".join(tokens).encode("utf-8")).hexdigest()[:32]
|
|
|
|
def _artifact(self, proposal, mandate, artifact_type: str, name: str, content: dict[str, Any], readable: str, generated_by: str, *, graph_run=None) -> VentureArtifact:
|
|
return VentureArtifact.objects.create(proposal=proposal, mandate=mandate, graph_run=graph_run, artifact_type=artifact_type, name=name, content=content, readable=readable, generated_by=generated_by)
|
|
|
|
def _readable_report(self, content: dict[str, Any]) -> str:
|
|
lines = ["# CRYPTO VENTURE COHORT V0.1 REPORT", "", f"Cohort ID: {content['cohort_id']}", f"Accepted protocols: {content['accepted_protocols']}", "", "## Ranking"]
|
|
for row in content["ranking"]:
|
|
lines.append(f"- Rank {row.get('rank')}: {row['company']} | {row['crypto_ic_decision']} | score {row['crypto_ic_score']} | token {row['token_necessity_classification']}")
|
|
lines.append("")
|
|
lines.append("## Top 3")
|
|
if content["top_3"]:
|
|
lines.extend(f"- {row['company']}" for row in content["top_3"])
|
|
else:
|
|
lines.append("Fewer than 3 qualified; weak token ideas were not promoted.")
|
|
lines.append("")
|
|
lines.append("Stop condition: no token sale, fundraising, mainnet issuance, user contact, or real spend.")
|
|
return "\n".join(lines)
|