daniel/Artifex
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Repair crypto protocol cohort V0.3.1 funnel

Browse source
This commit is contained in:
Daniel Maddern 2026-08-16 21:56:09 +07:00
parent d35a212aff
commit 554a865927
6 changed files with 249 additions and 107 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

272
agents/crypto_venture.py
View file

@ -58,22 +58,24 @@ from research.searxng import SearxngSearchClient, WebPageFetcher
CRYPTO_TERRITORIES = [ CRYPTO_TERRITORIES = [
"MACHINE_ECONOMIC_IDENTITY", "DECENTRALIZED_AI_COMPUTE",
"VERIFIABLE_INFERENCE_NETWORKS",
"MACHINE_SERVICE_MARKETS",
"AGENT_ECONOMIC_IDENTITY",
"AUTONOMOUS_AGENT_COMMERCE",
"DECENTRALIZED_SECURITY_MARKETS", "DECENTRALIZED_SECURITY_MARKETS",
"VERIFIABLE_AI_INFERENCE",
"PROOF_MARKETS",
"DECENTRALIZED_COMPUTE",
"DATA_PROVENANCE_NETWORKS",
"MACHINE_TO_MACHINE_SETTLEMENT",
"AUTONOMOUS_ESCROW",
"DECENTRALIZED_INSURANCE_RISK_POOLS",
"RESOURCE_RIGHTS_MARKETS",
"DECENTRALIZED_BANDWIDTH_STORAGE",
"AI_MODEL_CONTRIBUTION_NETWORKS",
"SOFTWARE_SECURITY_BONDS", "SOFTWARE_SECURITY_BONDS",
"AUTONOMOUS_AGENT_COORDINATION", "PROOF_MARKETS",
"DECENTRALIZED_ORACLE_ATTESTATION", "DECENTRALIZED_DATA_CONTRIBUTION",
"CRYPTO_NATIVE_CREATOR_DATA_RIGHTS", "MODEL_CONTRIBUTION_NETWORKS",
"COMPUTE_CONTRIBUTION_NETWORKS",
"DECENTRALIZED_RESOURCE_MARKETS",
"BANDWIDTH_STORAGE_RESOURCE_MARKETS",
"DECENTRALIZED_INSURANCE_RISK_POOLS",
"PROTOCOLIZED_ESCROW",
"MACHINE_REPUTATION_WITH_ECONOMIC_STAKE",
"PROVIDER_COORDINATION_NETWORKS",
"AUTONOMOUS_INFRASTRUCTURE_MARKETS",
"OPEN_CRYPTO_CATEGORY", "OPEN_CRYPTO_CATEGORY",
] ]
@ -111,7 +113,11 @@ CRYPTO_SCORE_DIMENSIONS = [
"PRE_TOKEN_MONETIZATION_POTENTIAL", "PRE_TOKEN_MONETIZATION_POTENTIAL",
] ]
CRYPTO_RESEARCH_CATEGORIES = ["USER_PAIN", "EXISTING_PROTOCOLS", "FAILED_PRECEDENTS", "ONCHAIN_ALTERNATIVES", "OFFCHAIN_ALTERNATIVES", "TOKEN_MODELS", "COLLATERAL_MODELS", "STAKING_SLASHING_PRECEDENTS", "PROVIDER_ECONOMICS", "NETWORK_BOOTSTRAP", "SECURITY_FAILURES", "LEGAL_REGULATORY", "PRE_TOKEN_MONETIZATION", "TOKEN_LAUNCH_PRECEDENTS"] CRYPTO_RESEARCH_CATEGORIES = ["USER_PAIN", "EXISTING_PROTOCOLS", "FAILED_PRECEDENTS", "TOKEN_UTILITY_PRECEDENTS", "SECURITY_BOND_PRECEDENTS", "SLASHING_PRECEDENTS", "RESOURCE_MARKET_PRECEDENTS", "PROVIDER_ECONOMICS", "NETWORK_BOOTSTRAP", "VALUE_CAPTURE", "TOKENOMICS_FAILURES", "SECURITY_INCIDENTS", "ONCHAIN_ALTERNATIVES", "OFFCHAIN_ALTERNATIVES", "LEGAL_REGULATORY", "PRE_TOKEN_MONETIZATION"]
STRUCTURAL_TOKEN_ROLES = {"SECURITY_BOND", "SLASHABLE_COLLATERAL", "RESOURCE_ALLOCATION", "MACHINE_ECONOMIC_IDENTITY", "PROVIDER_ADMISSION", "DECENTRALIZED_SUPPLY_COORDINATION"}
WEAK_TOKEN_ROLES = {"DEMAND_SIDE_PAYMENT", "GOVERNANCE", "PROVIDER_REWARD", "TREASURY", "ACCESS"}
TOKEN_UTILITY_SCORE_KEYS = ["SECURITY_UTILITY", "COORDINATION_UTILITY", "COLLATERAL_UTILITY", "SLASHING_UTILITY", "RESOURCE_ALLOCATION_UTILITY", "INCENTIVE_UTILITY", "NETWORK_BOOTSTRAP_UTILITY", "VALUE_CAPTURE_UTILITY", "MACHINE_ECONOMIC_UTILITY", "GOVERNANCE_UTILITY", "PAYMENT_UTILITY", "SPECULATION_DEPENDENCE"]
PRIMARY_SOURCE_DOMAINS = ["ethereum.org", "eips.ethereum.org", "github.com", "arxiv.org", "sec.gov", "cftc.gov", "immunefi.com"] PRIMARY_SOURCE_DOMAINS = ["ethereum.org", "eips.ethereum.org", "github.com", "arxiv.org", "sec.gov", "cftc.gov", "immunefi.com"]
SOURCE_REJECTION_REASONS = {"IRRELEVANT": "SOURCE_REJECT_IRRELEVANT", "WRONG_CATEGORY": "SOURCE_REJECT_WRONG_CATEGORY", "LOW_AUTHORITY": "SOURCE_REJECT_LOW_AUTHORITY", "SEARCH_NOISE": "SOURCE_REJECT_SEARCH_NOISE", "UNPARSABLE": "SOURCE_REJECT_UNPARSABLE", "DUPLICATE": "SOURCE_REJECT_DUPLICATE"} SOURCE_REJECTION_REASONS = {"IRRELEVANT": "SOURCE_REJECT_IRRELEVANT", "WRONG_CATEGORY": "SOURCE_REJECT_WRONG_CATEGORY", "LOW_AUTHORITY": "SOURCE_REJECT_LOW_AUTHORITY", "SEARCH_NOISE": "SOURCE_REJECT_SEARCH_NOISE", "UNPARSABLE": "SOURCE_REJECT_UNPARSABLE", "DUPLICATE": "SOURCE_REJECT_DUPLICATE"}
@ -148,15 +154,15 @@ class CryptoVentureService:
objective="Generate crypto-native protocol ventures where onchain execution and a native token are genuinely necessary for real usage, not speculation.", 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}, 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"], 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.3", "venture_track": VentureTrack.CRYPTO_PROTOCOL}, metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.3.1", "venture_track": VentureTrack.CRYPTO_PROTOCOL},
) )
self._artifact(None, mandate, "CRYPTO_VENTURE_MANDATE", "Crypto Protocol Venture V0.3 Mandate", mandate.constraints, "Crypto protocol cohort mandate. Sol-native problem-first ideation. No token sale, fundraising, or mainnet issuance.", "crypto_venture") self._artifact(None, mandate, "CRYPTO_VENTURE_MANDATE", "Crypto Protocol Venture V0.3.1 Mandate", mandate.constraints, "Crypto protocol cohort mandate. Sol-native problem-first ideation and strong native token utility search. No token sale, fundraising, or mainnet issuance.", "crypto_venture")
return mandate return mandate
def prepare_cohort(self, *, size: int, graph_run=None, concurrency: int = 2) -> VentureCohort: def prepare_cohort(self, *, size: int, graph_run=None, concurrency: int = 2) -> VentureCohort:
mandate = self.create_crypto_mandate() mandate = self.create_crypto_mandate()
cohort_id = f"CPV03-{time.strftime('%Y%m%d%H%M%S')}-{uuid.uuid4().hex[:8]}" cohort_id = f"CPV031-{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 v3"}, research_policy={"source_linked_evidence_required": True, "research_insufficient_if_no_sources": True, "primary_source_first": True, "minimum_category_evidence": "one strong source or multiple independent medium sources", "crypto_categories": CRYPTO_RESEARCH_CATEGORIES}, scoring_policy={"dimensions": CRYPTO_SCORE_DIMENSIONS, "strict_token_filter": True, "sol_problem_first_generation": True}, metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.3", "venture_track": VentureTrack.CRYPTO_PROTOCOL, "real_spend": 0, "real_customer_outreach": False, "token_sale": False, "fundraising": False, "mainnet_issuance": False, "v02_baseline": {"raw": 30, "crypto_survivors": 2, "autonomous_survivors": 0, "finalists": 0}}) 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 v3.1"}, research_policy={"source_linked_evidence_required": True, "research_insufficient_if_no_sources": True, "primary_source_first": True, "minimum_category_evidence": "one strong source or multiple independent medium sources", "crypto_categories": CRYPTO_RESEARCH_CATEGORIES}, scoring_policy={"dimensions": CRYPTO_SCORE_DIMENSIONS, "strict_token_filter": True, "sol_problem_first_generation": True, "strong_native_token_utility_required": True, "structural_roles": sorted(STRUCTURAL_TOKEN_ROLES)}, metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.3.1", "venture_track": VentureTrack.CRYPTO_PROTOCOL, "real_spend": 0, "real_customer_outreach": False, "token_sale": False, "fundraising": False, "mainnet_issuance": False, "v02_baseline": {"raw": 30, "crypto_survivors": 2, "autonomous_survivors": 0, "finalists": 0}, "v03_baseline": {"raw": 20, "onchain_rejected": 5, "onchain_passed": 15, "token_unnecessary": 12, "token_optional": 3, "crypto_survivors": 0, "finalists": 0}})
def generate_protocols(self, cohort: VentureCohort) -> list[CompanyProposal]: def generate_protocols(self, cohort: VentureCohort) -> list[CompanyProposal]:
accepted = [] accepted = []
@ -170,7 +176,7 @@ class CryptoVentureService:
proposal = self._create_proposal(cohort.mandate, payload, source, territory) proposal = self._create_proposal(cohort.mandate, payload, source, territory)
VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"territory": territory}) VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"territory": territory})
accepted.append(proposal) accepted.append(proposal)
cohort.metrics = {**cohort.metrics, "raw_target": cohort.cohort_size, "raw_generated": len(accepted), "raw_sol_theses": len([p for p in accepted if p.metadata.get("generation_source") == "sol"]), "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}), "model_usage": dict(model_usage), "sol_generation_requests": model_usage.get("sol_generation_requests", 0), "fallback_count": model_usage.get("fallback_count", 0)} cohort.metrics = {**cohort.metrics, "raw_theses": len(accepted), "raw_target": cohort.cohort_size, "raw_generated": len(accepted), "raw_sol_theses": len([p for p in accepted if p.metadata.get("generation_source") == "sol"]), "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}), "model_usage": dict(model_usage), "sol_generation_requests": model_usage.get("sol_generation_requests", 0), "sol_onchain_judge_requests": model_usage.get("sol_onchain_judge_requests", 0), "fallback_count": model_usage.get("fallback_count", 0)}
cohort.status = "PROTOCOLS_GENERATED" cohort.status = "PROTOCOLS_GENERATED"
cohort.save(update_fields=["metrics", "status", "updated_at"]) cohort.save(update_fields=["metrics", "status", "updated_at"])
return accepted return accepted
@ -185,10 +191,11 @@ class CryptoVentureService:
VentureGenerationRejection.objects.create(cohort=cohort, slot_index=member.rank or 0, attempt=1, decision=NoveltyGateDecision.REGENERATE_HARD_EXCLUSION, reason="removed by onchain necessity gate", candidate={"title": proposal.title, "onchain_necessity_score": assessment.onchain_necessity_score}, similarity_score=0.0) VentureGenerationRejection.objects.create(cohort=cohort, slot_index=member.rank or 0, attempt=1, decision=NoveltyGateDecision.REGENERATE_HARD_EXCLUSION, reason="removed by onchain necessity gate", candidate={"title": proposal.title, "onchain_necessity_score": assessment.onchain_necessity_score}, similarity_score=0.0)
member.delete() member.delete()
proposal.status = CompanyProposalStatus.REJECTED proposal.status = CompanyProposalStatus.REJECTED
proposal.metadata = {**proposal.metadata, "routed_to_saas": True, "onchain_rejected": True} proposal.metadata = {**proposal.metadata, "routed_to_saas": True, "routed_to_venture_studio": True, "onchain_rejected": True, "crypto_failure_reason": "onchain necessity failed"}
proposal.save(update_fields=["status", "metadata", "updated_at"]) proposal.save(update_fields=["status", "metadata", "updated_at"])
removed += 1 removed += 1
cohort.metrics = {**cohort.metrics, "onchain_rejected": cohort.metrics.get("onchain_rejected", 0) + removed, "crypto_survivors_after_onchain": cohort.members.count()} onchain_passed = cohort.members.count()
cohort.metrics = {**cohort.metrics, "onchain_rejected": cohort.metrics.get("onchain_rejected", 0) + removed, "onchain_passed": onchain_passed, "crypto_survivors_after_onchain": onchain_passed, "token_design_skipped_due_to_onchain_rejection": cohort.metrics.get("token_design_skipped_due_to_onchain_rejection", 0) + removed}
cohort.save(update_fields=["metrics", "updated_at"]) cohort.save(update_fields=["metrics", "updated_at"])
return assessments return assessments
@ -226,20 +233,56 @@ class CryptoVentureService:
def regenerate_rejected_slots(self, cohort: VentureCohort, *, reason: str = "novelty_or_token_gate") -> list[CompanyProposal]: def regenerate_rejected_slots(self, cohort: VentureCohort, *, reason: str = "novelty_or_token_gate") -> list[CompanyProposal]:
created: list[CompanyProposal] = [] created: list[CompanyProposal] = []
cohort.metrics = {**cohort.metrics, "regeneration_attempts": cohort.metrics.get("regeneration_attempts", 0), "regenerated_protocols": cohort.metrics.get("regenerated_protocols", 0), "crypto_survivors": cohort.members.count(), "unfilled_slots_after_regeneration": max(0, cohort.cohort_size - cohort.members.count()), "regeneration_policy": "V0.3 uses total raw Sol generation budget; offchain or weak token ideas are not force-filled."} cohort.metrics = {**cohort.metrics, "regeneration_attempts": cohort.metrics.get("regeneration_attempts", 0), "regenerated_protocols": cohort.metrics.get("regenerated_protocols", 0), "crypto_survivors": cohort.members.count(), "unfilled_slots_after_regeneration": max(0, cohort.cohort_size - cohort.members.count()), "regeneration_policy": "V0.3.1 uses total raw Sol generation budget; offchain or weak token ideas are not force-filled."}
cohort.save(update_fields=["metrics", "updated_at"]) cohort.save(update_fields=["metrics", "updated_at"])
return created return created
def token_necessity_gate(self, cohort: VentureCohort) -> list[TokenUtilityAssessment]: def token_necessity_gate(self, cohort: VentureCohort) -> list[TokenUtilityAssessment]:
assessments = [self.assess_token_necessity(member.proposal) for member in cohort.members.select_related("proposal")] model_usage = Counter(cohort.metrics.get("model_usage", {}))
before = cohort.members.count()
assessments = []
for member in cohort.members.select_related("proposal"):
if not member.proposal.metadata.get("crypto", {}).get("token_design_completed"):
used = self.derive_token_design(member.proposal)
model_usage.update(used)
assessments.append(self.assess_token_necessity(member.proposal))
model_usage["sol_token_utility_judge_requests"] += 1 if self.final_ic_model_hint == "sol" else 0
removed = self._remove_weak_token_members(cohort) removed = self._remove_weak_token_members(cohort)
cohort.metrics = {**cohort.metrics, **removed} cohort.metrics = {**cohort.metrics, **removed, "token_design_attempts": cohort.metrics.get("token_design_attempts", 0) + before, "token_design_count_matches_onchain_passed": (cohort.metrics.get("token_design_attempts", 0) + before) == cohort.metrics.get("onchain_passed", before), "model_usage": dict(model_usage), "sol_token_design_requests": model_usage.get("sol_token_design_requests", 0), "sol_token_utility_judge_requests": model_usage.get("sol_token_utility_judge_requests", 0)}
cohort.save(update_fields=["metrics", "updated_at"]) cohort.save(update_fields=["metrics", "updated_at"])
return assessments return assessments
def derive_token_design(self, proposal: CompanyProposal) -> Counter:
usage = Counter()
crypto = proposal.metadata.get("crypto", {})
passes = crypto.get("generation_passes", {}) if isinstance(crypto, dict) else {}
pass_a = passes.get("pass_a_problem_network", {}) if isinstance(passes, dict) else {}
pass_b = passes.get("pass_b_onchain", {}) if isinstance(passes, dict) else {}
pass_c_prompt = "PASS C - MINIMAL DECENTRALIZED PROTOCOL ARCHITECTURE. Return JSON with protocol_architecture, participant_model, verification_model, settlement_model, resource_allocation_model, dispute_model, value_capture_model, autonomous_operating_loop. No token design yet. Thesis: " + json.dumps({"problem": pass_a or proposal.pitch, "onchain_review": pass_b}, default=str)
architecture = self._model_json(self.generation_model_hint, pass_c_prompt, purpose=ModelCapability.REASONING)
pass_d_prompt = "PASS D - NATIVE TOKEN UTILITY DESIGN. Derive the minimum native token utility only if it performs strong structural economic or security functions. Explore SECURITY_BOND, SLASHABLE_COLLATERAL, PROVIDER_ADMISSION, RESOURCE_ALLOCATION, MACHINE_ECONOMIC_IDENTITY, CONTRIBUTION_ACCOUNTING, SECURITY_BUDGET, PROTOCOL_FEE_ASSET, PROVIDER_REWARD, DEMAND_SIDE_PAYMENT, GOVERNANCE, ACCESS, TREASURY, OTHER. Payment/governance/rewards/access alone are insufficient. Return JSON with token_thesis, token_utility list, token_demand_loop list, value_capture, token_roles object where each role is REQUIRED/STRONGLY_USEFUL/USEFUL/OPTIONAL/UNNECESSARY, mechanism_fingerprint object, security_model, regulatory_risks. Architecture: " + json.dumps(architecture or pass_a or proposal.pitch, default=str)
token_design = self._model_json(self.generation_model_hint, pass_d_prompt, purpose=ModelCapability.REASONING)
usage["sol_protocol_architecture_requests" if self.generation_model_hint == "sol" else "qwen_requests"] += 1
usage["sol_token_design_requests" if self.generation_model_hint == "sol" else "qwen_requests"] += 1
if token_design:
crypto = {**crypto, "generation_passes": {**passes, "pass_c_protocol_architecture": architecture, "pass_d_token_utility": token_design}, "token_design_completed": True, "token_roles": token_design.get("token_roles", {}), "mechanism_fingerprint": token_design.get("mechanism_fingerprint", {})}
proposal.metadata = {**proposal.metadata, "crypto": crypto}
proposal.pitch = {**proposal.pitch, "Token thesis": str(token_design.get("token_thesis", proposal.pitch.get("Token thesis", ""))), "Token demand loop": self._as_list(token_design.get("token_demand_loop") or proposal.pitch.get("Token demand loop", []))}
proposal.save(update_fields=["metadata", "pitch", "updated_at"])
if hasattr(proposal, "protocol_thesis"):
proposal.protocol_thesis.token_thesis = str(token_design.get("token_thesis", proposal.protocol_thesis.token_thesis))
proposal.protocol_thesis.utility_categories = self._as_list(token_design.get("token_utility") or proposal.protocol_thesis.utility_categories)
proposal.protocol_thesis.save(update_fields=["token_thesis", "utility_categories", "updated_at"])
if hasattr(proposal, "token_demand_loop"):
proposal.token_demand_loop.loop = self._as_list(token_design.get("token_demand_loop") or proposal.token_demand_loop.loop)
proposal.token_demand_loop.save(update_fields=["loop", "updated_at"])
return usage
def _remove_weak_token_members(self, cohort: VentureCohort) -> dict[str, int]: def _remove_weak_token_members(self, cohort: VentureCohort) -> dict[str, int]:
unnecessary = 0 unnecessary = 0
optional = 0 optional = 0
essential = 0
strongly = 0
for member in list(cohort.members.select_related("proposal")): for member in list(cohort.members.select_related("proposal")):
proposal = member.proposal proposal = member.proposal
assessment = getattr(proposal, "token_utility_assessment", None) or self.assess_token_necessity(proposal) assessment = getattr(proposal, "token_utility_assessment", None) or self.assess_token_necessity(proposal)
@ -251,9 +294,13 @@ class CryptoVentureService:
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) 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() member.delete()
proposal.status = CompanyProposalStatus.REJECTED proposal.status = CompanyProposalStatus.REJECTED
proposal.metadata = {**proposal.metadata, "routed_to_saas": assessment.classification == TokenNecessityClassification.TOKEN_OPTIONAL} proposal.metadata = {**proposal.metadata, "routed_to_saas": assessment.classification == TokenNecessityClassification.TOKEN_OPTIONAL, "routed_to_venture_studio": True, "crypto_failure_reason": f"token classification {assessment.classification}"}
proposal.save(update_fields=["status", "metadata", "updated_at"]) 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, "crypto_survivors": cohort.members.count()} elif assessment.classification == TokenNecessityClassification.TOKEN_ESSENTIAL:
essential += 1
else:
strongly += 1
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, "token_essential": cohort.metrics.get("token_essential", 0) + essential, "token_strongly_justified": cohort.metrics.get("token_strongly_justified", 0) + strongly, "serious_crypto_survivors": cohort.members.count(), "crypto_survivors": cohort.members.count()}
def light_market_research(self, cohort: VentureCohort) -> None: def light_market_research(self, cohort: VentureCohort) -> None:
started = time.monotonic() started = time.monotonic()
@ -442,9 +489,13 @@ class CryptoVentureService:
return "UNAVAILABLE" return "UNAVAILABLE"
def token_red_team(self, cohort: VentureCohort) -> list[CryptoRedTeamAssessment]: def token_red_team(self, cohort: VentureCohort) -> list[CryptoRedTeamAssessment]:
assessments = [self.red_team_proposal(member.proposal) for member in cohort.members.select_related("proposal")] assessments = []
for member in cohort.members.select_related("proposal"):
token = getattr(member.proposal, "token_utility_assessment", None)
if token and token.classification in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED}:
assessments.append(self.red_team_proposal(member.proposal))
counts = Counter(flag for item in assessments for flag in item.flags) counts = Counter(flag for item in assessments for flag in item.flags)
cohort.metrics = {**cohort.metrics, "token_red_team_failures": dict(counts)} cohort.metrics = {**cohort.metrics, "token_red_team_failures": dict(counts), "token_red_team_count": len(assessments), "sol_token_red_team_requests": len(assessments) if self.final_ic_model_hint == "sol" else 0}
cohort.save(update_fields=["metrics", "updated_at"]) cohort.save(update_fields=["metrics", "updated_at"])
return assessments return assessments
@ -492,10 +543,10 @@ class CryptoVentureService:
row["rank"] = rank 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}}) 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}})
model_usage = Counter(cohort.metrics.get("model_usage", {})) model_usage = Counter(cohort.metrics.get("model_usage", {}))
model_usage["sol_judge_requests"] += len([row for row in rows if row.get("token_necessity_classification")]) model_usage["sol_token_utility_judge_requests"] = max(model_usage.get("sol_token_utility_judge_requests", 0), cohort.metrics.get("sol_token_utility_judge_requests", 0))
model_usage["sol_final_ic_requests"] += len(rows) if self.final_ic_model_hint == "sol" else 0 model_usage["sol_final_ic_requests"] += len(rows) if self.final_ic_model_hint == "sol" else 0
model_usage["qwen_requests"] += 0 model_usage["qwen_requests"] += 0
cohort.metrics = {**cohort.metrics, "crypto_ranked_count": len(rows), "crypto_top_3_count": len(top_3), "autonomous_crypto_survivors": len([row for row in rows if row.get("crypto_autonomy_class") == CryptoAutonomyClass.AUTONOMOUS_CRYPTO]), "finalists": len(top_3), "model_usage": dict(model_usage), "sol_judge_requests": model_usage.get("sol_judge_requests", 0), "sol_final_ic_requests": model_usage.get("sol_final_ic_requests", 0), "qwen_requests": model_usage.get("qwen_requests", 0)} cohort.metrics = {**cohort.metrics, "crypto_ranked_count": len(rows), "crypto_top_3_count": len(top_3), "autonomous_crypto_survivors": len([row for row in rows if row.get("crypto_autonomy_class") == CryptoAutonomyClass.AUTONOMOUS_CRYPTO]), "assisted_high_potential": len([row for row in rows if row.get("crypto_ic_decision") == CryptoICDecisionType.ASSISTED_CRYPTO_HIGH_POTENTIAL]), "finalists": len(top_3), "model_usage": dict(model_usage), "sol_token_utility_judge_requests": model_usage.get("sol_token_utility_judge_requests", 0), "sol_final_ic_requests": model_usage.get("sol_final_ic_requests", 0), "qwen_requests": model_usage.get("qwen_requests", 0)}
cohort.status = "CRYPTO_IC_COMPLETE" cohort.status = "CRYPTO_IC_COMPLETE"
cohort.save(update_fields=["metrics", "status", "updated_at"]) cohort.save(update_fields=["metrics", "status", "updated_at"])
return review return review
@ -586,9 +637,10 @@ class CryptoVentureService:
rows = review.rankings rows = review.rankings
research_health = self._research_health(rows) research_health = self._research_health(rows)
routed_to_saas = [{"company": proposal.title, "token_classification": getattr(proposal, "token_utility_assessment", None).classification if hasattr(proposal, "token_utility_assessment") else None, "product_thesis": proposal.description, "reason": "Failed onchain/token counterfactual but may be valuable as pre-token/SaaS product."} for proposal in cohort.mandate.company_proposals.filter(status=CompanyProposalStatus.REJECTED, metadata__routed_to_saas=True).order_by("created_at")[:10]] routed_to_saas = [{"company": proposal.title, "token_classification": getattr(proposal, "token_utility_assessment", None).classification if hasattr(proposal, "token_utility_assessment") else None, "product_thesis": proposal.description, "reason": "Failed onchain/token counterfactual but may be valuable as pre-token/SaaS product."} for proposal in cohort.mandate.company_proposals.filter(status=CompanyProposalStatus.REJECTED, metadata__routed_to_saas=True).order_by("created_at")[:10]]
comparison = {"v02": cohort.metadata.get("v02_baseline", {"raw": 30, "crypto_survivors": 2, "autonomous_survivors": 0, "finalists": 0}), "v03": {"raw": cohort.metrics.get("raw_generated", 0), "crypto_survivors": cohort.members.count(), "autonomous_survivors": cohort.metrics.get("autonomous_crypto_survivors", 0), "finalists": len(review.top_3), "average_research_quality": research_health.get("average_coverage", 0), "idea_diversity": review.concentration}} comparison = {"v02": cohort.metadata.get("v02_baseline", {"raw": 30, "crypto_survivors": 2, "autonomous_survivors": 0, "finalists": 0}), "v03": cohort.metadata.get("v03_baseline", {}), "v031": {"raw": cohort.metrics.get("raw_generated", 0), "onchain_passed": cohort.metrics.get("onchain_passed", 0), "onchain_rejected": cohort.metrics.get("onchain_rejected", 0), "token_strongly_justified": cohort.metrics.get("token_strongly_justified", 0), "token_essential": cohort.metrics.get("token_essential", 0), "crypto_survivors": cohort.members.count(), "autonomous_survivors": cohort.metrics.get("autonomous_crypto_survivors", 0), "assisted_high_potential": cohort.metrics.get("assisted_high_potential", 0), "finalists": len(review.top_3), "average_research_quality": research_health.get("average_coverage", 0), "idea_diversity": review.concentration}}
content = {"title": "CRYPTO VENTURE COHORT V0.3", "cohort_id": cohort.cohort_id, "graph_run": str(cohort.graph_run_id or ""), "runtime": cohort.metrics, "raw_protocols_generated": cohort.metrics.get("raw_generated", 0), "raw_sol_theses": cohort.metrics.get("raw_sol_theses", 0), "onchain_rejected": cohort.metrics.get("onchain_rejected", 0), "token_unnecessary": cohort.metrics.get("token_unnecessary_rejections", 0), "token_optional_routed_saas": cohort.metrics.get("token_optional_route_to_saas", 0), "duplicates": cohort.metrics.get("duplicate_rejections", 0), "crypto_survivors": cohort.members.count(), "autonomous_crypto_survivors": cohort.metrics.get("autonomous_crypto_survivors", 0), "security_blocked": cohort.metrics.get("protocol_security_gate_blocked", 0), "finalists": len(review.top_3), "generation_attempts": cohort.metrics.get("generation_attempts", 0), "model_usage": cohort.metrics.get("model_usage", {}), "ranking": rows, "top_3": review.top_3, "routed_to_saas": routed_to_saas, "v02_comparison": comparison, "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, "research_health": research_health, "system_metrics": cohort.metrics, "stop_conditions": {"token_sale": False, "nft_sale": False, "founding_membership_sale": False, "fundraising": False, "mainnet_issuance": False, "user_contact": False, "real_spend": 0}} security_summary = {"guard_reviews": cohort.metrics.get("protocol_security_gate_passed", 0) + cohort.metrics.get("protocol_security_gate_blocked", 0), "pass": cohort.metrics.get("protocol_security_gate_passed", 0), "revise_or_pass": cohort.metrics.get("protocol_security_gate_passed", 0), "block": cohort.metrics.get("protocol_security_gate_blocked", 0)}
artifact = self._artifact(None, cohort.mandate, "CRYPTO_VENTURE_COHORT_REPORT", "Crypto Venture Cohort V0.3 Report", content, self._readable_report(content), "crypto_portfolio_ic", graph_run=cohort.graph_run) content = {"title": "CRYPTO VENTURE COHORT V0.3.1", "cohort_id": cohort.cohort_id, "graph_run": str(cohort.graph_run_id or ""), "runtime": cohort.metrics, "raw_protocols_generated": cohort.metrics.get("raw_generated", 0), "raw_sol_theses": cohort.metrics.get("raw_sol_theses", 0), "onchain_rejected": cohort.metrics.get("onchain_rejected", 0), "onchain_passed": cohort.metrics.get("onchain_passed", 0), "token_design_attempts": cohort.metrics.get("token_design_attempts", 0), "token_design_skipped": cohort.metrics.get("token_design_skipped_due_to_onchain_rejection", 0), "token_unnecessary": cohort.metrics.get("token_unnecessary_rejections", 0), "token_optional_routed_saas": cohort.metrics.get("token_optional_route_to_saas", 0), "token_strongly_justified": cohort.metrics.get("token_strongly_justified", 0), "token_essential": cohort.metrics.get("token_essential", 0), "duplicates": cohort.metrics.get("duplicate_rejections", 0), "crypto_survivors": cohort.members.count(), "autonomous_crypto_survivors": cohort.metrics.get("autonomous_crypto_survivors", 0), "assisted_high_potential": cohort.metrics.get("assisted_high_potential", 0), "security_blocked": cohort.metrics.get("protocol_security_gate_blocked", 0), "finalists": len(review.top_3), "generation_attempts": cohort.metrics.get("generation_attempts", 0), "model_usage": cohort.metrics.get("model_usage", {}), "research_summary": research_health, "security_summary": security_summary, "ranking": rows, "top_3": review.top_3, "routed_to_saas": routed_to_saas, "v02_v03_v031_comparison": comparison, "v02_comparison": comparison, "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, "research_health": research_health, "system_metrics": cohort.metrics, "stop_conditions": {"token_sale": False, "nft_sale": False, "founding_membership_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.3.1 Report", content, self._readable_report(content), "crypto_portfolio_ic", graph_run=cohort.graph_run)
cohort.status = "COMPLETE" cohort.status = "COMPLETE"
cohort.save(update_fields=["status", "updated_at"]) cohort.save(update_fields=["status", "updated_at"])
return artifact return artifact
@ -613,24 +665,28 @@ class CryptoVentureService:
def assess_token_necessity(self, proposal: CompanyProposal) -> TokenUtilityAssessment: def assess_token_necessity(self, proposal: CompanyProposal) -> TokenUtilityAssessment:
crypto = proposal.metadata.get("crypto", {}) crypto = proposal.metadata.get("crypto", {})
text = self._proposal_text(proposal) text = self._proposal_text(proposal)
utilities = [u for u in crypto.get("token_utility", []) if isinstance(u, str)] or self._infer_utilities(text) utilities = [u for u in crypto.get("token_utility", []) if isinstance(u, str)] or self._as_list(getattr(proposal, "protocol_thesis", None).utility_categories if hasattr(proposal, "protocol_thesis") else []) or self._infer_utilities(text)
roles = self.token_role_decomposition(proposal) roles = self.token_role_decomposition(proposal)
utility_profile = self.token_utility_profile(proposal, roles)
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"])) 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"]))
required_roles = len(roles.required_roles) strong_roles = [role for role, value in roles.roles.items() if value in {"REQUIRED", "STRONGLY_USEFUL"}]
structural_roles = [role for role in strong_roles if role in STRUCTURAL_TOKEN_ROLES]
required_roles = len(strong_roles)
weak = any(term in text for term in ["meme", "speculative", "community token", "marketing token", "community marketing", "governance token only", "token gated subscription"]) weak = any(term in text for term in ["meme", "speculative", "community token", "marketing token", "community marketing", "governance token only", "token gated subscription"])
payment_only = roles.required_roles == ["PAYMENT"] weak_only = bool(strong_roles) and not structural_roles and all(role in WEAK_TOKEN_ROLES for role in strong_roles)
score = min(100, 25 + strong_count * 10 + required_roles * 16 + (12 if "slash" in text or "slashing" in text else 0) + (8 if "provider" in text else 0) - (25 if payment_only else 0) - (35 if weak else 0)) payment_only = strong_roles == ["DEMAND_SIDE_PAYMENT"] or roles.roles.get("PROTOCOL_FEE_ASSET") in {"REQUIRED", "STRONGLY_USEFUL"} and len(strong_roles) <= 1
score = min(100, 35 + strong_count * 4 + len(strong_roles) * 8 + len(structural_roles) * 13 + (12 if "slash" in text or "slashing" in text else 0) + (8 if "provider" in text else 0) - (35 if payment_only or weak_only else 0) - (35 if weak else 0) - int(utility_profile.get("SPECULATION_DEPENDENCE", 0) * 0.25))
economic_model = getattr(proposal, "token_economic_model", None) economic_model = getattr(proposal, "token_economic_model", None)
model_payload = {"unit_of_service": economic_model.unit_of_service, "payment_asset": economic_model.payment_asset, "stake_requirement": economic_model.stake_requirement, "collateral_requirement": economic_model.collateral_requirement} if economic_model else {} model_payload = {"unit_of_service": economic_model.unit_of_service, "payment_asset": economic_model.payment_asset, "stake_requirement": economic_model.stake_requirement, "collateral_requirement": economic_model.collateral_requirement} if economic_model else {}
sol_review = self._sol_json("CRYPTO_TOKEN_COUNTERFACTUAL_JUDGE V0.3. Independent from generation. Compare MODEL A native token, MODEL B stablecoin payments + native bond token, MODEL C stablecoin payments + ETH/stablecoin collateral, MODEL D onchain system with no proprietary token, MODEL E centralized SaaS/database. Return JSON with classification TOKEN_ESSENTIAL, TOKEN_STRONGLY_JUSTIFIED, TOKEN_OPTIONAL, or TOKEN_UNNECESSARY; score 0-100; rationale; fiat_or_database_substitution; model_a_native_token; model_b_stablecoin_payment_native_bond; model_c_external_collateral; model_d_no_proprietary_token; model_e_centralized_saas; minimum_required_crypto_architecture; minimum_required_token_architecture; native_token_removed_breaks list; native_token_removed_degradation NONE/MINOR/MODERATE/MATERIAL/CRITICAL; native_token_removed_explanation. Product/protocol/token/economic model: " + json.dumps({"pitch": proposal.pitch, "crypto": crypto, "economic_model": model_payload}, default=str)) sol_review = self._sol_json("CRYPTO_TOKEN_UTILITY_JUDGE V0.3.1. Independent from ideation, onchain review, and token design. Challenge the token; do not merely summarize it. A native token need not be mathematically impossible to substitute, but benefits must outweigh costs. Evaluate security quality, attack economics, slashing effectiveness, provider skin-in-the-game, capital efficiency, protocol-specific risk pricing, permissionless supply formation, participant coordination, resource allocation, reputation portability, value capture, network effects, bootstrapping economics, governance, and composability versus USDC/ETH/external collateral/non-transferable attestations. Return JSON with token_classification TOKEN_ESSENTIAL/TOKEN_STRONGLY_JUSTIFIED/TOKEN_OPTIONAL/TOKEN_UNNECESSARY; score 0-100; utility_scores object; strong_structural_roles list; weak_roles list; counterfactual_analysis; material_improvements_over_USDC_ETH list; native_asset_costs_and_risks list; argument_for_native_token; argument_against_native_token; final_rationale; native_token_removed_breaks list; native_token_removed_degradation NONE/MINOR/MODERATE/MATERIAL/CRITICAL; native_token_removed_explanation. Product/protocol/token/economic model: " + json.dumps({"pitch": proposal.pitch, "crypto": crypto, "roles": roles.roles, "utility_profile": utility_profile, "economic_model": model_payload}, default=str))
if sol_review: if sol_review:
score = float(sol_review.get("score", score)) score = float(sol_review.get("score", score))
sol_classification = str(sol_review.get("classification", "")) sol_classification = str(sol_review.get("token_classification") or sol_review.get("classification", ""))
else: else:
sol_classification = "" sol_classification = ""
if score >= 82: if len(strong_roles) >= 2 and structural_roles and score >= 85:
classification = TokenNecessityClassification.TOKEN_ESSENTIAL classification = TokenNecessityClassification.TOKEN_ESSENTIAL
elif score >= 70: elif len(strong_roles) >= 2 and structural_roles and score >= 75:
classification = TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED classification = TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED
elif score >= 50: elif score >= 50:
classification = TokenNecessityClassification.TOKEN_OPTIONAL classification = TokenNecessityClassification.TOKEN_OPTIONAL
@ -638,43 +694,75 @@ class CryptoVentureService:
classification = TokenNecessityClassification.TOKEN_UNNECESSARY classification = TokenNecessityClassification.TOKEN_UNNECESSARY
if sol_classification in TokenNecessityClassification.values: if sol_classification in TokenNecessityClassification.values:
classification = sol_classification classification = sol_classification
degradation = str((sol_review or {}).get("native_token_removed_degradation", "MATERIAL" if required_roles else "MINOR")) degradation = str((sol_review or {}).get("native_token_removed_degradation", "MATERIAL" if structural_roles else "MINOR"))
if degradation in {"NONE", "MINOR"} and classification in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED}: if weak:
classification = TokenNecessityClassification.TOKEN_UNNECESSARY
score = min(score, 35)
elif payment_only or weak_only or not structural_roles:
classification = TokenNecessityClassification.TOKEN_OPTIONAL if score >= 45 else TokenNecessityClassification.TOKEN_UNNECESSARY
score = min(score, 62)
if (len(strong_roles) < 2 or not structural_roles or degradation in {"NONE", "MINOR"}) and classification in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED}:
classification = TokenNecessityClassification.TOKEN_OPTIONAL classification = TokenNecessityClassification.TOKEN_OPTIONAL
score = min(score, 62) score = min(score, 62)
proposal.metadata = {**proposal.metadata, "crypto": {**crypto, "counterfactual_judge_model": self.final_ic_model_hint, "counterfactual_judge_used": bool(sol_review)}} proposal.metadata = {**proposal.metadata, "crypto": {**crypto, "token_utility_judge_model": self.final_ic_model_hint, "token_utility_judge_used": bool(sol_review), "token_utility_profile": utility_profile, "strong_structural_roles": structural_roles, "strong_token_roles": strong_roles}}
proposal.save(update_fields=["metadata", "updated_at"]) proposal.save(update_fields=["metadata", "updated_at"])
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 derives from genuinely required roles, not native payment currency alone.")), "metadata": {"weak_utility_terms": INSUFFICIENT_TOKEN_UTILITIES, "sol_counterfactual_review": sol_review or {}, "independent_counterfactual_judge": "CRYPTO_TOKEN_COUNTERFACTUAL_JUDGE", "required_roles": roles.required_roles, "payment_only_penalty": payment_only, "model_a_native_token": (sol_review or {}).get("model_a_native_token", {}), "model_b_stablecoin_payment_native_bond": (sol_review or {}).get("model_b_stablecoin_payment_native_bond", {}), "model_c_external_collateral": (sol_review or {}).get("model_c_external_collateral", {}), "model_d_no_proprietary_token": (sol_review or {}).get("model_d_no_proprietary_token", {}), "model_e_centralized_saas": (sol_review or {}).get("model_e_centralized_saas", {}), "minimum_required_crypto_architecture": (sol_review or {}).get("minimum_required_crypto_architecture", ""), "minimum_required_token_architecture": (sol_review or {}).get("minimum_required_token_architecture", ""), "native_token_removed_breaks": self._as_list((sol_review or {}).get("native_token_removed_breaks")), "native_token_removed_degradation": degradation, "native_token_removed_explanation": (sol_review or {}).get("native_token_removed_explanation", "")}})[0] 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("counterfactual_analysis", "USDC/ETH/external collateral may preserve basic operation but must be compared against protocol-specific security and coordination improvements.")), "rationale": str((sol_review or {}).get("final_rationale") or (sol_review or {}).get("rationale") or "Token score derives from structural utility, not native payment currency alone."), "metadata": {"weak_utility_terms": INSUFFICIENT_TOKEN_UTILITIES, "sol_counterfactual_review": sol_review or {}, "sol_token_utility_judge": sol_review or {}, "independent_token_utility_judge": "CRYPTO_TOKEN_UTILITY_JUDGE", "argument_for_native_token": (sol_review or {}).get("argument_for_native_token", ""), "argument_against_native_token": (sol_review or {}).get("argument_against_native_token", ""), "utility_scores": (sol_review or {}).get("utility_scores", utility_profile), "strong_structural_roles": structural_roles, "strong_token_roles": strong_roles, "weak_roles": (sol_review or {}).get("weak_roles", []), "material_improvements_over_USDC_ETH": self._as_list((sol_review or {}).get("material_improvements_over_USDC_ETH")), "native_asset_costs_and_risks": self._as_list((sol_review or {}).get("native_asset_costs_and_risks")), "required_roles": roles.required_roles, "payment_only_penalty": payment_only, "native_token_removed_breaks": self._as_list((sol_review or {}).get("native_token_removed_breaks")), "native_token_removed_degradation": degradation, "native_token_removed_explanation": (sol_review or {}).get("native_token_removed_explanation", "")}})[0]
def token_role_decomposition(self, proposal: CompanyProposal) -> TokenRoleDecomposition: def token_role_decomposition(self, proposal: CompanyProposal) -> TokenRoleDecomposition:
text = self._proposal_text(proposal) text = self._proposal_text(proposal)
explicit_roles = proposal.metadata.get("crypto", {}).get("token_roles", {}) if isinstance(proposal.metadata, dict) else {}
roles = {} roles = {}
role_terms = { role_terms = {
"PAYMENT": ["fee", "payment", "settlement"],
"SECURITY_BOND": ["bond", "security budget"], "SECURITY_BOND": ["bond", "security budget"],
"SLASHABLE_COLLATERAL": ["slash", "slashing", "collateral"], "SLASHABLE_COLLATERAL": ["slash", "slashing", "collateral"],
"PROVIDER_ADMISSION": ["provider admission", "permissionless provider", "admission"],
"RESOURCE_ALLOCATION": ["resource allocation", "scarce", "quota", "bandwidth", "storage", "compute slot"],
"MACHINE_ECONOMIC_IDENTITY": ["machine economic", "machine identity", "agent identity", "reputation"],
"CONTRIBUTION_ACCOUNTING": ["contribution", "contributor", "accounting"],
"SECURITY_BUDGET": ["security budget"],
"PROTOCOL_FEE_ASSET": ["protocol fee", "fee asset"],
"PROVIDER_REWARD": ["provider reward", "reward"],
"DEMAND_SIDE_PAYMENT": ["payment", "settlement"],
"GOVERNANCE": ["governance", "vote"], "GOVERNANCE": ["governance", "vote"],
"RESOURCE_ACCESS": ["access", "scarce", "quota"], "ACCESS": ["access"],
"REPUTATION": ["reputation", "attestation"],
"REWARD": ["reward", "provider reward"],
"TREASURY": ["treasury"], "TREASURY": ["treasury"],
"OTHER": [], "OTHER": [],
} }
for role, terms in role_terms.items(): for role, terms in role_terms.items():
if role == "PAYMENT" and any(term in text for term in terms): explicit = str(explicit_roles.get(role, "")).upper() if isinstance(explicit_roles, dict) else ""
value = TokenRoleRequirement.USEFUL if explicit in {"REQUIRED", "STRONGLY_USEFUL", "USEFUL", "OPTIONAL", "UNNECESSARY"}:
value = explicit
elif explicit_roles:
value = TokenRoleRequirement.UNNECESSARY
elif role == "GOVERNANCE" and any(term in text for term in terms) and not any(term in text for term in ["parameter", "slashing", "treasury"]): elif role == "GOVERNANCE" and any(term in text for term in terms) and not any(term in text for term in ["parameter", "slashing", "treasury"]):
value = TokenRoleRequirement.OPTIONAL value = TokenRoleRequirement.OPTIONAL
elif any(term in text for term in terms): elif any(term in text for term in terms):
value = TokenRoleRequirement.REQUIRED if role in {"SECURITY_BOND", "SLASHABLE_COLLATERAL", "REPUTATION"} else TokenRoleRequirement.USEFUL value = "REQUIRED" if role in STRUCTURAL_TOKEN_ROLES else "USEFUL"
else: else:
value = TokenRoleRequirement.UNNECESSARY value = TokenRoleRequirement.UNNECESSARY
roles[role] = value roles[role] = value
required = [role for role, value in roles.items() if value == TokenRoleRequirement.REQUIRED] required = [role for role, value in roles.items() if value in {"REQUIRED", "STRONGLY_USEFUL"}]
stablecoin = {"MODEL_A_native_payment_staking_governance": "baseline proposal", "MODEL_B_stablecoin_payment_native_bond": "preferred if payment utility is separable", "MODEL_C_stablecoin_payment_stablecoin_collateral": "valid if slashing/collateral does not need protocol-native exposure", "MODEL_D_onchain_no_proprietary_token": "valid if proprietary token adds no security/resource allocation advantage", "MODEL_E_centralized_saas_database": "routes to SaaS if verification/settlement/reputation do not degrade", "usdc_identical_payment_utility": roles.get("PAYMENT") != TokenRoleRequirement.REQUIRED, "external_collateral_identical_security": "SLASHABLE_COLLATERAL" not in required and "SECURITY_BOND" not in required, "stable_collateral_identical_security": "SLASHABLE_COLLATERAL" not in required and "SECURITY_BOND" not in required} stablecoin = {"MODEL_A_native_payment_staking_governance": "baseline proposal", "MODEL_B_stablecoin_payment_native_bond": "preferred if payment utility is separable", "MODEL_C_stablecoin_payment_stablecoin_collateral": "valid if slashing/collateral does not need protocol-native exposure", "MODEL_D_onchain_no_proprietary_token": "valid if proprietary token adds no security/resource allocation advantage", "MODEL_E_centralized_saas_database": "routes to SaaS if verification/settlement/reputation do not degrade", "usdc_identical_payment_utility": roles.get("DEMAND_SIDE_PAYMENT") not in {"REQUIRED", "STRONGLY_USEFUL"}, "external_collateral_identical_security": "SLASHABLE_COLLATERAL" not in required and "SECURITY_BOND" not in required, "stable_collateral_identical_security": "SLASHABLE_COLLATERAL" not in required and "SECURITY_BOND" not in required}
outcome = "Native token removal materially degrades security/reputation if required roles remain." if required else "Native token removed changes little; route to SaaS or revise token model." outcome = "Native token removal materially degrades security/reputation if required roles remain." if required else "Native token removed changes little; route to SaaS or revise token model."
return TokenRoleDecomposition.objects.update_or_create(proposal=proposal, defaults={"roles": roles, "required_roles": required, "stablecoin_counterfactual": stablecoin, "native_token_removed_outcome": outcome})[0] return TokenRoleDecomposition.objects.update_or_create(proposal=proposal, defaults={"roles": roles, "required_roles": required, "stablecoin_counterfactual": stablecoin, "native_token_removed_outcome": outcome})[0]
def token_utility_profile(self, proposal: CompanyProposal, roles: TokenRoleDecomposition) -> dict[str, int]:
text = self._proposal_text(proposal)
score = {key: 0 for key in TOKEN_UTILITY_SCORE_KEYS}
score["SECURITY_UTILITY"] = 85 if roles.roles.get("SECURITY_BOND") in {"REQUIRED", "STRONGLY_USEFUL"} else 35
score["COLLATERAL_UTILITY"] = 85 if roles.roles.get("SLASHABLE_COLLATERAL") in {"REQUIRED", "STRONGLY_USEFUL"} else 30
score["SLASHING_UTILITY"] = 82 if "slash" in text and roles.roles.get("SLASHABLE_COLLATERAL") in {"REQUIRED", "STRONGLY_USEFUL"} else 25
score["RESOURCE_ALLOCATION_UTILITY"] = 82 if roles.roles.get("RESOURCE_ALLOCATION") in {"REQUIRED", "STRONGLY_USEFUL"} else 25
score["MACHINE_ECONOMIC_UTILITY"] = 84 if roles.roles.get("MACHINE_ECONOMIC_IDENTITY") in {"REQUIRED", "STRONGLY_USEFUL"} else 25
score["COORDINATION_UTILITY"] = 75 if any(roles.roles.get(role) in {"REQUIRED", "STRONGLY_USEFUL"} for role in ["PROVIDER_ADMISSION", "CONTRIBUTION_ACCOUNTING", "RESOURCE_ALLOCATION"]) else 35
score["INCENTIVE_UTILITY"] = 70 if roles.roles.get("PROVIDER_REWARD") in {"REQUIRED", "STRONGLY_USEFUL", "USEFUL"} else 25
score["NETWORK_BOOTSTRAP_UTILITY"] = 75 if "bootstrap" in text or roles.roles.get("PROVIDER_ADMISSION") in {"REQUIRED", "STRONGLY_USEFUL"} else 30
score["VALUE_CAPTURE_UTILITY"] = 75 if "value" in text or roles.roles.get("PROTOCOL_FEE_ASSET") in {"REQUIRED", "STRONGLY_USEFUL", "USEFUL"} else 30
score["GOVERNANCE_UTILITY"] = 45 if roles.roles.get("GOVERNANCE") in {"REQUIRED", "STRONGLY_USEFUL", "USEFUL"} else 10
score["PAYMENT_UTILITY"] = 45 if roles.roles.get("DEMAND_SIDE_PAYMENT") in {"REQUIRED", "STRONGLY_USEFUL", "USEFUL"} else 10
score["SPECULATION_DEPENDENCE"] = 85 if any(term in text for term in ["price appreciation", "buyback", "yield", "apy", "speculat"]) else 15
return score
def assess_pretoken_monetization(self, proposal: CompanyProposal) -> PreTokenMonetizationAssessment: def assess_pretoken_monetization(self, proposal: CompanyProposal) -> PreTokenMonetizationAssessment:
text = self._proposal_text(proposal) text = self._proposal_text(proposal)
mode = PreTokenMonetizationMode.STABLECOIN_USAGE_FEES if "usage" in text or "api" in text else PreTokenMonetizationMode.PAID_BETA_ACCESS if "testnet" in text else PreTokenMonetizationMode.NONE mode = PreTokenMonetizationMode.STABLECOIN_USAGE_FEES if "usage" in text or "api" in text else PreTokenMonetizationMode.PAID_BETA_ACCESS if "testnet" in text else PreTokenMonetizationMode.NONE
@ -711,6 +799,9 @@ class CryptoVentureService:
if payload.get("generation_passes"): if payload.get("generation_passes"):
proposal.metadata = {**proposal.metadata, "crypto": {**proposal.metadata["crypto"], "generation_passes": payload.get("generation_passes", {}), "problem_first_pass_a_had_no_token": True, "pre_token_product": payload.get("pre_token_product"), "pre_token_customer": payload.get("pre_token_customer"), "pre_token_value": payload.get("pre_token_value"), "pre_token_payment_method": payload.get("pre_token_payment_method"), "pre_token_revenue_model": payload.get("pre_token_revenue_model")}} proposal.metadata = {**proposal.metadata, "crypto": {**proposal.metadata["crypto"], "generation_passes": payload.get("generation_passes", {}), "problem_first_pass_a_had_no_token": True, "pre_token_product": payload.get("pre_token_product"), "pre_token_customer": payload.get("pre_token_customer"), "pre_token_value": payload.get("pre_token_value"), "pre_token_payment_method": payload.get("pre_token_payment_method"), "pre_token_revenue_model": payload.get("pre_token_revenue_model")}}
proposal.save(update_fields=["metadata", "updated_at"]) proposal.save(update_fields=["metadata", "updated_at"])
if payload.get("token_roles"):
proposal.metadata = {**proposal.metadata, "crypto": {**proposal.metadata["crypto"], "token_roles": payload.get("token_roles", {})}}
proposal.save(update_fields=["metadata", "updated_at"])
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) 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"]) 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}) self.bus.publish("CRYPTO_PROTOCOL_PROPOSED", payload={"proposal_id": str(proposal.id), "territory": territory, "source": source})
@ -727,10 +818,7 @@ class CryptoVentureService:
pass_b_prompt = "PASS B - ONCHAIN NECESSITY. Independently evaluate whether this coordination system materially improves when implemented onchain. Return JSON with onchain_necessity_score 0-100, why_onchain, offchain_substitute, pass_onchain boolean, reasons list. Consider trust minimization, neutral settlement, permissionless participation, verifiable state, escrow, programmable guarantees, censorship resistance, cross-organization coordination, and machine-to-machine interaction. Thesis: " + json.dumps(pass_a, default=str) pass_b_prompt = "PASS B - ONCHAIN NECESSITY. Independently evaluate whether this coordination system materially improves when implemented onchain. Return JSON with onchain_necessity_score 0-100, why_onchain, offchain_substitute, pass_onchain boolean, reasons list. Consider trust minimization, neutral settlement, permissionless participation, verifiable state, escrow, programmable guarantees, censorship resistance, cross-organization coordination, and machine-to-machine interaction. Thesis: " + json.dumps(pass_a, default=str)
pass_b = self._model_json(self.generation_model_hint, pass_b_prompt, purpose=ModelCapability.REASONING) pass_b = self._model_json(self.generation_model_hint, pass_b_prompt, purpose=ModelCapability.REASONING)
usage["sol_onchain_judge_requests" if self.generation_model_hint == "sol" else "qwen_requests"] += 1 usage["sol_onchain_judge_requests" if self.generation_model_hint == "sol" else "qwen_requests"] += 1
pass_c_prompt = "PASS C - DERIVED TOKEN DESIGN. Only if the onchain system needs a native token, derive the minimum token architecture from required protocol functions. Prefer native security bond, endogenous slashable collateral, scarce resource allocation, decentralized security budget, permissionless provider admission, machine economic identity, protocol-specific risk collateral, or required decentralized supply incentives. Weak roles such as payment, discounts, governance, rewards, community, token-gating, treasury, marketing, or alignment cannot justify the token alone. Return JSON with token_thesis, token_utility list, token_demand_loop list, value_capture, security_model, regulatory_risks, required_token_functions, weak_token_functions. If no native token is needed, return token_thesis='No native token required' and token_utility=[]. Problem thesis: " + json.dumps(pass_a, default=str) + " Onchain review: " + json.dumps(pass_b, default=str) payload = {**pass_a, **pass_b, "token_thesis": "Native token not assessed yet; token utility design runs only after onchain qualification.", "token_utility": [], "token_demand_loop": [], "user_pain": pass_a.get("coordination_failure") or pass_a.get("trust_problem"), "protocol_thesis": pass_a.get("why_neutral_network_may_help") or pass_a.get("product_thesis"), "business_model": pass_a.get("pre_token_revenue_model"), "fee_model": pass_a.get("pre_token_payment_method"), "generation_passes": {"pass_a_problem_network": pass_a, "pass_b_onchain": pass_b}}
pass_c = self._model_json(self.generation_model_hint, pass_c_prompt, purpose=ModelCapability.REASONING)
usage["sol_token_design_requests" if self.generation_model_hint == "sol" else "qwen_requests"] += 1
payload = {**pass_a, **pass_b, **pass_c, "user_pain": pass_a.get("coordination_failure") or pass_a.get("trust_problem"), "protocol_thesis": pass_a.get("why_neutral_network_may_help") or pass_a.get("product_thesis"), "business_model": pass_a.get("pre_token_revenue_model"), "fee_model": pass_a.get("pre_token_payment_method"), "generation_passes": {"pass_a_problem_network": pass_a, "pass_b_onchain": pass_b, "pass_c_token": pass_c}}
return payload, self.generation_model_hint, usage return payload, self.generation_model_hint, usage
except Exception: except Exception:
pass pass
@ -762,44 +850,43 @@ class CryptoVentureService:
roles = self.token_role_decomposition(proposal) roles = self.token_role_decomposition(proposal)
flags = [] flags = []
if token.classification in {TokenNecessityClassification.TOKEN_OPTIONAL, TokenNecessityClassification.TOKEN_UNNECESSARY}: if token.classification in {TokenNecessityClassification.TOKEN_OPTIONAL, TokenNecessityClassification.TOKEN_UNNECESSARY}:
flags.append(TokenRedTeamFlag.TOKEN_NOT_REQUIRED) flags.append("TOKEN_NOT_REQUIRED")
if onchain.onchain_necessity_score < 70: if onchain.onchain_necessity_score < 70:
flags.append(TokenRedTeamFlag.ONCHAIN_NOT_REQUIRED) flags.append("REJECT_ONCHAIN_NOT_NEEDED")
if value.value_accrual_quality_score < 65: if value.value_accrual_quality_score < 65:
flags.append(TokenRedTeamFlag.VALUE_CAPTURE_BROKEN) flags.append("VALUE_CAPTURE_WEAK")
if value.tokenomics_sustainability_score < 65: if value.tokenomics_sustainability_score < 65:
flags.append(TokenRedTeamFlag.UNSUSTAINABLE_EMISSIONS) flags.append("UNSUSTAINABLE_EMISSIONS")
if policy.regulatory_manageability_score < 55: if policy.regulatory_manageability_score < 55:
flags.append(TokenRedTeamFlag.REGULATORY_RISK_HIGH) flags.append("LEGAL_COMPLEXITY_OUTWEIGHS_UTILITY")
text = self._proposal_text(proposal) text = self._proposal_text(proposal)
if "speculat" in text or "apy" in text or "yield farm" in text: if "speculat" in text or "apy" in text or "yield farm" in text:
flags.append(TokenRedTeamFlag.SPECULATION_DEPENDENT) flags.append("SPECULATION_DEPENDENT")
if "governance" in text and "slash" not in text and "fee" not in text: if "governance" in text and "slash" not in text and "fee" not in text:
flags.append(TokenRedTeamFlag.GOVERNANCE_THEATER) flags.append("GOVERNANCE_ONLY_TOKEN")
if len([role for role, value in roles.roles.items() if value in {TokenRoleRequirement.REQUIRED, TokenRoleRequirement.USEFUL}]) >= 6: if len([role for role, value in roles.roles.items() if value in {TokenRoleRequirement.REQUIRED, TokenRoleRequirement.USEFUL}]) >= 6:
flags.append(TokenRedTeamFlag.TOKEN_DOES_TOO_MUCH) flags.append("TOKEN_DOES_TOO_MUCH")
if roles.roles.get("PAYMENT") in {TokenRoleRequirement.USEFUL, TokenRoleRequirement.OPTIONAL}: if roles.roles.get("DEMAND_SIDE_PAYMENT") in {"USEFUL", "OPTIONAL", "REQUIRED", "STRONGLY_USEFUL"} and not any(roles.roles.get(role) in {"REQUIRED", "STRONGLY_USEFUL"} for role in STRUCTURAL_TOKEN_ROLES):
flags.append(TokenRedTeamFlag.NATIVE_PAYMENT_NOT_REQUIRED) flags.append("PAYMENT_ONLY_TOKEN")
flags.append("NATIVE_PAYMENT_NOT_REQUIRED")
if roles.stablecoin_counterfactual.get("stable_collateral_identical_security"): if roles.stablecoin_counterfactual.get("stable_collateral_identical_security"):
flags.append(TokenRedTeamFlag.STABLECOIN_SUBSTITUTE_WORKS) flags.append("COLLATERAL_UTILITY_WEAK")
flags.append(TokenRedTeamFlag.COLLATERAL_CAN_BE_EXTERNAL)
if any(term in text for term in ["buyback", "revenue share", "profit share"]): if any(term in text for term in ["buyback", "revenue share", "profit share"]):
flags.append(TokenRedTeamFlag.REVENUE_CLAIM_LANGUAGE) flags.append("REVENUE_CLAIM_LANGUAGE")
if "buyback" in text: if "buyback" in text:
flags.append(TokenRedTeamFlag.BUYBACK_DEPENDENCY) flags.append("BUYBACK_DEPENDENCY")
if "yield" in text or "apy" in text: if "yield" in text or "apy" in text:
flags.append(TokenRedTeamFlag.YIELD_DEPENDENCY) flags.append("YIELD_DEPENDENCY")
if "manual dispute" in text or "human court" in text: if "manual dispute" in text or "human court" in text:
flags.append(TokenRedTeamFlag.HUMAN_DISPUTE_DEPENDENCY) flags.append("SLASHING_DEPENDS_ON_HUMAN_JUDGMENT")
if "central operator" in text: if "central operator" in text:
flags.append(TokenRedTeamFlag.CENTRAL_OPERATOR_CONTRADICTION) flags.append("CENTRAL_OPERATOR_CONTRADICTION")
if "subjective" in text and "slashing" in text: if "subjective" in text and "slashing" in text:
flags.append(TokenRedTeamFlag.SLASHING_NOT_OBJECTIVELY_MEASURABLE) flags.append("SLASHING_NOT_OBJECTIVE")
sol_review = self._sol_json("Independent Token Red Team V0.2. 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, TOKEN_DOES_TOO_MUCH, NATIVE_PAYMENT_NOT_REQUIRED, STABLECOIN_SUBSTITUTE_WORKS, COLLATERAL_CAN_BE_EXTERNAL, REVENUE_CLAIM_LANGUAGE, BUYBACK_DEPENDENCY, YIELD_DEPENDENCY, SPECULATIVE_BOOTSTRAP, EMISSIONS_DEPENDENCY, SECURITY_TOKEN_CONCENTRATION, SLASHING_NOT_OBJECTIVELY_MEASURABLE, ORACLE_CIRCULARITY, HUMAN_DISPUTE_DEPENDENCY, CENTRAL_OPERATOR_CONTRADICTION, NETWORK_NOT_READY_FOR_DECENTRALIZATION; severity LOW/MEDIUM/HIGH; critique. Proposal: " + json.dumps(self.crypto_score_context(proposal), default=str)) sol_review = self._sol_json("Independent Token Red Team V0.3.1. Return JSON with flags list using only TOKEN_DOES_TOO_MUCH, PAYMENT_ONLY_TOKEN, GOVERNANCE_ONLY_TOKEN, REWARD_ONLY_TOKEN, NATIVE_PAYMENT_NOT_REQUIRED, COLLATERAL_UTILITY_WEAK, SLASHING_NOT_OBJECTIVE, SLASHING_DEPENDS_ON_HUMAN_JUDGMENT, TOKEN_SECURITY_NOT_LINKED_TO_PROTOCOL_RISK, SPECULATION_DEPENDENT, REVENUE_CLAIM_LANGUAGE, BUYBACK_DEPENDENCY, YIELD_DEPENDENCY, UNSUSTAINABLE_EMISSIONS, MERCENARY_PROVIDER_INCENTIVES, RESOURCE_ALLOCATION_ARTIFICIAL, MACHINE_IDENTITY_NOT_PORTABLE, VALUE_CAPTURE_WEAK, BOOTSTRAP_REQUIRES_TOKEN_PRICE_GROWTH, CENTRAL_OPERATOR_CONTRADICTION, LEGAL_COMPLEXITY_OUTWEIGHS_UTILITY; severity LOW/MEDIUM/HIGH; critique. Proposal: " + json.dumps(self.crypto_score_context(proposal), default=str))
if sol_review: if sol_review:
for flag in self._as_list(sol_review.get("flags")): for flag in self._as_list(sol_review.get("flags")):
if str(flag) in TokenRedTeamFlag.values: flags.append(str(flag))
flags.append(str(flag))
severity = "HIGH" if len(flags) >= 3 else "MEDIUM" if flags else "LOW" 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"}: if sol_review and str(sol_review.get("severity", "")) in {"LOW", "MEDIUM", "HIGH"}:
severity = str(sol_review["severity"]) severity = str(sol_review["severity"])
@ -841,7 +928,7 @@ class CryptoVentureService:
roles = getattr(proposal, "token_role_decomposition", None) or self.token_role_decomposition(proposal) roles = getattr(proposal, "token_role_decomposition", None) or self.token_role_decomposition(proposal)
base_users = max(1, model.demand_side_users) base_users = max(1, model.demand_side_users)
tx_per_user = 10 tx_per_user = 10
scenario_multipliers = {"LOW_USAGE": 0.3, "EXPECTED": 1.0, "HIGH_USAGE": 3.0, "NO_SUBSIDY": 0.9, "TOKEN_PRICE_DOWN_80": 0.8, "TOKEN_PRICE_UP_10X": 1.2, "PROVIDER_CHURN_50": 0.7, "USER_GROWTH_10X": 10.0, "HIGH_SLASH_RATE": 1.0, "ZERO_EMISSIONS": 1.0, "STABLECOIN_PAYMENT": 1.0, "EXTERNAL_COLLATERAL": 1.0, "NATIVE_TOKEN_REMOVED": 1.0} scenario_multipliers = {"LOW_USAGE": 0.3, "EXPECTED_USAGE": 1.0, "HIGH_USAGE": 3.0, "NO_SUBSIDY": 0.9, "TOKEN_PRICE_DOWN_80": 0.8, "TOKEN_PRICE_UP_10X": 1.2, "PROVIDER_CHURN_50": 0.7, "USER_GROWTH_10X": 10.0, "HIGH_SLASH_RATE": 1.0, "ZERO_EMISSIONS": 1.0, "STABLECOIN_PAYMENT": 1.0, "ETH_COLLATERAL": 1.0, "USDC_COLLATERAL": 1.0, "NATIVE_TOKEN_REMOVED": 1.0}
for name, multiplier in scenario_multipliers.items(): for name, multiplier in scenario_multipliers.items():
users = int(base_users * multiplier) users = int(base_users * multiplier)
tx = int(users * tx_per_user) tx = int(users * tx_per_user)
@ -854,10 +941,11 @@ class CryptoVentureService:
provider_supply = max(1, int(provider_supply * 0.45)) provider_supply = max(1, int(provider_supply * 0.45))
staking = max(100, int(staking * 0.45)) staking = max(100, int(staking * 0.45))
native_removed = name == "NATIVE_TOKEN_REMOVED" native_removed = name == "NATIVE_TOKEN_REMOVED"
stablecoin_substitute = name in {"STABLECOIN_PAYMENT", "EXTERNAL_COLLATERAL"} stablecoin_substitute = name in {"STABLECOIN_PAYMENT", "ETH_COLLATERAL", "USDC_COLLATERAL"}
protocol_degrades = native_removed and bool(roles.required_roles) degradation_dimensions = {"SECURITY_DEGRADATION": "SECURITY_BOND" in roles.required_roles or "SLASHABLE_COLLATERAL" in roles.required_roles, "COORDINATION_DEGRADATION": "PROVIDER_ADMISSION" in roles.required_roles, "RESOURCE_ALLOCATION_DEGRADATION": "RESOURCE_ALLOCATION" in roles.required_roles, "BOOTSTRAP_DEGRADATION": "DECENTRALIZED_SUPPLY_COORDINATION" in roles.required_roles, "VALUE_CAPTURE_DEGRADATION": "PROTOCOL_FEE_ASSET" in roles.required_roles, "NETWORK_EFFECT_DEGRADATION": bool(roles.required_roles), "MACHINE_ECONOMIC_DEGRADATION": "MACHINE_ECONOMIC_IDENTITY" in roles.required_roles}
scenarios[name] = {"users": users, "transactions": tx, "fees": fees, "token_demand": 0 if native_removed else fees + staking * 0.01, "provider_supply": provider_supply, "staking": 0 if native_removed else staking, "emissions": emissions, "treasury": round(1000 + fees * model.treasury_share - emissions * 0.1, 2), "circulating_supply": 0 if native_removed else 1_000_000 + emissions, "token_velocity": 0 if native_removed else round(tx / max(1, fees + staking * 0.01), 2), "reward_coverage": round(fees / max(1, emissions), 2), "network_security_budget": 0 if native_removed else staking, "slash_events": round(tx * slash_rate, 2), "stablecoin_payment_works": stablecoin_substitute or roles.roles.get("PAYMENT") != TokenRoleRequirement.REQUIRED, "external_collateral_works": name == "EXTERNAL_COLLATERAL" and roles.stablecoin_counterfactual.get("external_collateral_identical_security"), "protocol_degrades_without_native_token": protocol_degrades} protocol_degrades = native_removed and any(degradation_dimensions.values())
summary = {"token_price_appreciation_primary_success_variable": False, "subsidy_removed_survives": scenarios["NO_SUBSIDY"]["reward_coverage"] >= 1.0, "native_token_removed_degrades_protocol": scenarios["NATIVE_TOKEN_REMOVED"]["protocol_degrades_without_native_token"], "stablecoin_payment_substitute_works": scenarios["STABLECOIN_PAYMENT"]["stablecoin_payment_works"], "external_collateral_substitute_works": scenarios["EXTERNAL_COLLATERAL"]["external_collateral_works"], "stress_notes": ["Price appreciation is not a success variable.", "If native-token-removed changes little, reduce token necessity."]} scenarios[name] = {"users": users, "transactions": tx, "fees": fees, "token_demand": 0 if native_removed else fees + staking * 0.01, "provider_supply": provider_supply, "staking": 0 if native_removed else staking, "emissions": emissions, "treasury": round(1000 + fees * model.treasury_share - emissions * 0.1, 2), "circulating_supply": 0 if native_removed else 1_000_000 + emissions, "token_velocity": 0 if native_removed else round(tx / max(1, fees + staking * 0.01), 2), "reward_coverage": round(fees / max(1, emissions), 2), "network_security_budget": 0 if native_removed else staking, "slash_events": round(tx * slash_rate, 2), "stablecoin_payment_works": stablecoin_substitute or roles.roles.get("DEMAND_SIDE_PAYMENT") not in {"REQUIRED", "STRONGLY_USEFUL"}, "external_collateral_works": name in {"ETH_COLLATERAL", "USDC_COLLATERAL"} and roles.stablecoin_counterfactual.get("external_collateral_identical_security"), "protocol_degrades_without_native_token": protocol_degrades, "native_token_removed_degradation_dimensions": degradation_dimensions if native_removed else {}}
summary = {"token_price_appreciation_primary_success_variable": False, "subsidy_removed_survives": scenarios["NO_SUBSIDY"]["reward_coverage"] >= 1.0, "native_token_removed_degrades_protocol": scenarios["NATIVE_TOKEN_REMOVED"]["protocol_degrades_without_native_token"], "native_token_removed_degradation_dimensions": scenarios["NATIVE_TOKEN_REMOVED"]["native_token_removed_degradation_dimensions"], "stablecoin_payment_substitute_works": scenarios["STABLECOIN_PAYMENT"]["stablecoin_payment_works"], "eth_collateral_substitute_works": scenarios["ETH_COLLATERAL"]["external_collateral_works"], "usdc_collateral_substitute_works": scenarios["USDC_COLLATERAL"]["external_collateral_works"], "stress_notes": ["Price appreciation is not a success variable.", "Native-token-removed asks whether protocol becomes materially worse, not literally impossible."]}
return TokenomicsSimulation.objects.update_or_create(proposal=proposal, defaults={"scenarios": scenarios, "summary": summary})[0] return TokenomicsSimulation.objects.update_or_create(proposal=proposal, defaults={"scenarios": scenarios, "summary": summary})[0]
def crypto_scenario_lab(self, cohort: VentureCohort) -> list[CryptoScenarioLab]: def crypto_scenario_lab(self, cohort: VentureCohort) -> list[CryptoScenarioLab]:
@ -878,6 +966,7 @@ class CryptoVentureService:
roles = getattr(proposal, "token_role_decomposition", None) or self.token_role_decomposition(proposal) roles = getattr(proposal, "token_role_decomposition", None) or self.token_role_decomposition(proposal)
security = getattr(proposal, "protocol_security_assessment", None) security = getattr(proposal, "protocol_security_assessment", None)
autonomy = self.assess_autonomy(proposal) autonomy = self.assess_autonomy(proposal)
utility_profile = token.metadata.get("utility_scores", proposal.metadata.get("crypto", {}).get("token_utility_profile", {})) if isinstance(token.metadata, dict) else {}
text = self._proposal_text(proposal) text = self._proposal_text(proposal)
component_scores = { component_scores = {
"TOKEN_NECESSITY": token.token_necessity_score, "TOKEN_NECESSITY": token.token_necessity_score,
@ -896,7 +985,7 @@ class CryptoVentureService:
decision = self._crypto_decision(component_scores, token.classification, red.flags) decision = self._crypto_decision(component_scores, token.classification, red.flags)
autonomy_class = CryptoAutonomyClass.AUTONOMOUS_CRYPTO if autonomy.autonomous_operability_score >= 70 else CryptoAutonomyClass.ASSISTED_CRYPTO autonomy_class = CryptoAutonomyClass.AUTONOMOUS_CRYPTO if autonomy.autonomous_operability_score >= 70 else CryptoAutonomyClass.ASSISTED_CRYPTO
counterfactual = token.metadata.get("sol_counterfactual_review", {}) if isinstance(token.metadata, dict) else {} counterfactual = token.metadata.get("sol_counterfactual_review", {}) if isinstance(token.metadata, dict) else {}
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, "user": proposal.target_customer, "pre_token_business": pretoken.pre_token_business, "pre_token_monetization": pretoken.mode, "pre_token_monetization_potential": pretoken.pre_token_monetization_potential, "pre_token_revenue_ladder": pretoken.revenue_ladder, "protocol_economy": proposal.protocol_thesis.protocol_thesis, "token_economy": proposal.protocol_thesis.token_thesis, "token_thesis": proposal.protocol_thesis.token_thesis, "why_onchain": proposal.onchain_assessment.rationale, "why_token": token.rationale, "token_role_decomposition": roles.roles, "stablecoin_counterfactual": roles.stablecoin_counterfactual, "external_collateral_counterfactual": counterfactual.get("model_c_external_collateral", {}), "native_token_removed_breaks": token.metadata.get("native_token_removed_breaks", []), "native_token_removed_degradation": token.metadata.get("native_token_removed_degradation", ""), "native_token_removed_explanation": token.metadata.get("native_token_removed_explanation", ""), "native_token_removed_outcome": roles.native_token_removed_outcome, "native_token_removed_degrades_protocol": sim.summary.get("native_token_removed_degrades_protocol", False), "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, "crypto_autonomy_class": autonomy_class, "regulatory_manageability": policy.regulatory_manageability_score, "security_assessment": security.decision if security else "NOT_ASSESSED", "security_risk": component_scores["SECURITY_MODEL_QUALITY"], "guard_verdict": security.decision if security else "NOT_ASSESSED", "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", {}), "research_confidence": proposal.metadata.get("research", {}).get("research_status", "RESEARCH_INSUFFICIENT"), "legal_review_required": policy.legal_review_required} 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, "user": proposal.target_customer, "pre_token_business": pretoken.pre_token_business, "pre_token_monetization": pretoken.mode, "pre_token_monetization_potential": pretoken.pre_token_monetization_potential, "pre_token_revenue_ladder": pretoken.revenue_ladder, "protocol_economy": proposal.protocol_thesis.protocol_thesis, "token_economy": proposal.protocol_thesis.token_thesis, "token_thesis": proposal.protocol_thesis.token_thesis, "why_onchain": proposal.onchain_assessment.rationale, "why_token": token.rationale, "token_role_decomposition": roles.roles, "token_utility_scores": utility_profile, "strong_token_roles": token.metadata.get("strong_token_roles", []), "strong_structural_roles": token.metadata.get("strong_structural_roles", []), "argument_for_native_token": token.metadata.get("argument_for_native_token", ""), "argument_against_native_token": token.metadata.get("argument_against_native_token", ""), "stablecoin_counterfactual": roles.stablecoin_counterfactual, "external_collateral_counterfactual": counterfactual.get("model_c_external_collateral", {}), "native_token_removed_breaks": token.metadata.get("native_token_removed_breaks", []), "native_token_removed_degradation": token.metadata.get("native_token_removed_degradation", ""), "native_token_removed_explanation": token.metadata.get("native_token_removed_explanation", ""), "native_token_removed_outcome": roles.native_token_removed_outcome, "native_token_removed_degrades_protocol": sim.summary.get("native_token_removed_degrades_protocol", False), "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, "crypto_autonomy_class": autonomy_class, "regulatory_manageability": policy.regulatory_manageability_score, "security_assessment": security.decision if security else "NOT_ASSESSED", "security_risk": component_scores["SECURITY_MODEL_QUALITY"], "guard_verdict": security.decision if security else "NOT_ASSESSED", "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", {}), "research_confidence": proposal.metadata.get("research", {}).get("research_status", "RESEARCH_INSUFFICIENT"), "legal_review_required": policy.legal_review_required}
def assess_autonomy(self, proposal: CompanyProposal) -> AutonomousOperabilityAssessment: def assess_autonomy(self, proposal: CompanyProposal) -> AutonomousOperabilityAssessment:
assessment = self.venture.assess_autonomous_operability(proposal) assessment = self.venture.assess_autonomous_operability(proposal)
@ -922,19 +1011,21 @@ class CryptoVentureService:
if classification == TokenNecessityClassification.TOKEN_UNNECESSARY: if classification == TokenNecessityClassification.TOKEN_UNNECESSARY:
return CryptoICDecisionType.REJECT_TOKEN_NOT_NEEDED return CryptoICDecisionType.REJECT_TOKEN_NOT_NEEDED
if classification == TokenNecessityClassification.TOKEN_OPTIONAL: if classification == TokenNecessityClassification.TOKEN_OPTIONAL:
return CryptoICDecisionType.ROUTE_TO_SAAS return CryptoICDecisionType.REJECT_TOKEN_UTILITY_WEAK
if TokenRedTeamFlag.SPECULATION_DEPENDENT in flags: if TokenRedTeamFlag.SPECULATION_DEPENDENT in flags:
return CryptoICDecisionType.REJECT_SPECULATIVE return CryptoICDecisionType.REJECT_SPECULATIVE
if scores["REGULATORY_MANAGEABILITY"] < 55: if scores["REGULATORY_MANAGEABILITY"] < 55:
return CryptoICDecisionType.REJECT_REGULATORY_RISK return CryptoICDecisionType.REJECT_REGULATORY_RISK
if scores["TOKENOMICS_SUSTAINABILITY"] < 65 or scores["VALUE_ACCRUAL_QUALITY"] < 65: if scores["TOKENOMICS_SUSTAINABILITY"] < 65 or scores["VALUE_ACCRUAL_QUALITY"] < 65:
return CryptoICDecisionType.REVISE_TOKEN_MODEL return CryptoICDecisionType.REVISE_TOKEN_MODEL
if scores["TOKEN_NECESSITY"] >= 85 and scores["ONCHAIN_NECESSITY"] >= 80 and scores["REAL_USAGE_DEMAND"] >= 75 and 55 <= scores["AUTONOMOUS_OPERABILITY"] < 70:
return CryptoICDecisionType.ASSISTED_CRYPTO_HIGH_POTENTIAL
if scores["TOKEN_NECESSITY"] >= 82 and scores["ONCHAIN_NECESSITY"] >= 75: if scores["TOKEN_NECESSITY"] >= 82 and scores["ONCHAIN_NECESSITY"] >= 75:
return CryptoICDecisionType.TESTNET_PILOT return CryptoICDecisionType.TESTNET_PILOT
return CryptoICDecisionType.PROTOCOL_VALIDATE return CryptoICDecisionType.PROTOCOL_VALIDATE
def _sol_final_ic_adjustment(self, row: dict[str, Any]) -> dict[str, Any]: def _sol_final_ic_adjustment(self, row: dict[str, Any]) -> dict[str, Any]:
review = self._sol_json("Final Crypto IC review V0.3. Consider original Sol thesis, independent token counterfactual, Token Red Team, Guard assessment, token economic model, simulation, primary-source evidence, autonomy analysis, pre-token revenue model, and uncertainty. Return JSON with optional score_adjustment -20..20, decision from TESTNET_PILOT, PROTOCOL_VALIDATE, REVISE_TOKEN_MODEL, ROUTE_TO_SAAS, ROUTE_TO_ASSISTED_CRYPTO, WATCHLIST, REJECT_TOKEN_NOT_NEEDED, REJECT_ONCHAIN_NOT_NEEDED, REJECT_SECURITY_MODEL, REJECT_ECONOMIC_MODEL, REJECT_AUTONOMY, REJECT_NO_DEMAND, and rationale. Do not promote weak token ideas or fill finalist slots. Row: " + json.dumps(row, default=str)) review = self._sol_json("Final Crypto IC review V0.3.1. Consider problem thesis, protocol thesis, token utility profile, independent token judge, Token Red Team, Guard assessment, token economic model, simulation, primary-source evidence, autonomy analysis, pre-token business, and uncertainty. Return JSON with optional score_adjustment -20..20, decision from TESTNET_PILOT, PROTOCOL_VALIDATE, REVISE_TOKEN_MODEL, ASSISTED_CRYPTO_HIGH_POTENTIAL, WATCHLIST, ROUTE_TO_VENTURE_STUDIO, REJECT_TOKEN_NOT_NEEDED, REJECT_TOKEN_UTILITY_WEAK, REJECT_ONCHAIN_NOT_NEEDED, REJECT_SECURITY_MODEL, REJECT_ECONOMIC_MODEL, REJECT_AUTONOMY, REJECT_NO_DEMAND, and rationale. Do not promote weak token ideas or fill finalist slots. Row: " + json.dumps(row, default=str))
if not review: if not review:
return row return row
adjustment = max(-20, min(20, float(review.get("score_adjustment", 0)))) adjustment = max(-20, min(20, float(review.get("score_adjustment", 0))))
@ -945,7 +1036,8 @@ class CryptoVentureService:
def _qualifies_finalist(self, row: dict[str, Any]) -> bool: def _qualifies_finalist(self, row: dict[str, Any]) -> bool:
scores = row["component_scores"] 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 and row.get("security_assessment") != ProtocolSecurityDecision.BLOCK_TESTNET and row.get("native_token_removed_degrades_protocol") is True structural_roles = set(row.get("strong_structural_roles", []))
return row["token_necessity_classification"] in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED} and len(row.get("strong_token_roles", [])) >= 2 and structural_roles & STRUCTURAL_TOKEN_ROLES 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 and row.get("security_assessment") != ProtocolSecurityDecision.BLOCK_TESTNET and row.get("native_token_removed_degrades_protocol") is True
def _crypto_concentration(self, rows: list[dict[str, Any]]) -> dict[str, Any]: def _crypto_concentration(self, rows: list[dict[str, Any]]) -> dict[str, Any]:
categories = Counter(row["company"].split()[0] for row in rows) categories = Counter(row["company"].split()[0] for row in rows)
@ -1021,7 +1113,7 @@ class CryptoVentureService:
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) 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: def _readable_report(self, content: dict[str, Any]) -> str:
lines = ["# CRYPTO VENTURE COHORT V0.3", "", f"Cohort ID: {content['cohort_id']}", f"Crypto survivors: {content['crypto_survivors']}", "", "## Ranking"] lines = ["# CRYPTO VENTURE COHORT V0.3.1", "", f"Cohort ID: {content['cohort_id']}", f"Crypto survivors: {content['crypto_survivors']}", "", "## Ranking"]
for row in content["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(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("")

9
control_plane/ventures/management/commands/export_crypto_venture_cohort.py
View file

@ -11,7 +11,7 @@ from control_plane.ventures.models import VentureCohort
class Command(BaseCommand): class Command(BaseCommand):
help = "Export a Crypto Venture Cohort V0.3 report." help = "Export a Crypto Venture Cohort V0.3.1 report."
def add_arguments(self, parser): def add_arguments(self, parser):
parser.add_argument("--cohort", help="Cohort stable ID or primary key. Defaults to latest crypto cohort.") parser.add_argument("--cohort", help="Cohort stable ID or primary key. Defaults to latest crypto cohort.")
@ -64,9 +64,10 @@ class Command(BaseCommand):
f"Cohort ID: `{payload['cohort_id']}`", f"Cohort ID: `{payload['cohort_id']}`",
f"GraphRun: `{payload['graph_run']}`", f"GraphRun: `{payload['graph_run']}`",
f"Raw protocols generated: `{payload.get('raw_protocols_generated', payload.get('accepted_protocols', 0))}`", f"Raw protocols generated: `{payload.get('raw_protocols_generated', payload.get('accepted_protocols', 0))}`",
f"Raw Sol theses: `{payload.get('raw_sol_theses', payload.get('raw_protocols_generated', 0))}`. Onchain rejected: `{payload.get('onchain_rejected', 0)}`.", f"Raw Sol theses: `{payload.get('raw_sol_theses', payload.get('raw_protocols_generated', 0))}`. Onchain passed: `{payload.get('onchain_passed', 0)}`. Onchain rejected: `{payload.get('onchain_rejected', 0)}`.",
f"Token unnecessary: `{payload.get('token_unnecessary', 0)}`. Token optional/routed SaaS: `{payload.get('token_optional_routed_saas', 0)}`. Duplicates: `{payload.get('duplicates', 0)}`.", f"Token design attempts: `{payload.get('token_design_attempts', 0)}`. Token design skipped: `{payload.get('token_design_skipped', 0)}`.",
f"Crypto survivors: `{payload.get('crypto_survivors', 0)}`. Autonomous crypto survivors: `{payload.get('autonomous_crypto_survivors', 0)}`. Security-blocked: `{payload.get('security_blocked', 0)}`. Finalists: `{payload.get('finalists', len(payload.get('top_3', [])))}`.", f"Token unnecessary: `{payload.get('token_unnecessary', 0)}`. Token optional/routed SaaS: `{payload.get('token_optional_routed_saas', 0)}`. Token strongly justified: `{payload.get('token_strongly_justified', 0)}`. Token essential: `{payload.get('token_essential', 0)}`. Duplicates: `{payload.get('duplicates', 0)}`.",
f"Crypto survivors: `{payload.get('crypto_survivors', 0)}`. Autonomous crypto survivors: `{payload.get('autonomous_crypto_survivors', 0)}`. Assisted high-potential: `{payload.get('assisted_high_potential', 0)}`. Security-blocked: `{payload.get('security_blocked', 0)}`. Finalists: `{payload.get('finalists', len(payload.get('top_3', [])))}`.",
f"Generation attempts: `{payload['generation_attempts']}`", f"Generation attempts: `{payload['generation_attempts']}`",
"", "",
"## Runtime", "## Runtime",

8
control_plane/ventures/management/commands/run_crypto_venture_cohort.py
View file

@ -16,7 +16,7 @@ from model_router.router import ModelRouter
class Command(BaseCommand): class Command(BaseCommand):
help = "Run a Crypto / Protocol Venture Cohort V0.3." help = "Run a Crypto / Protocol Venture Cohort V0.3.1."
def add_arguments(self, parser): def add_arguments(self, parser):
parser.add_argument("--size", type=int, default=20) parser.add_argument("--size", type=int, default=20)
@ -90,10 +90,16 @@ class Command(BaseCommand):
"raw_protocols_generated": cohort.metrics.get("raw_generated", 0), "raw_protocols_generated": cohort.metrics.get("raw_generated", 0),
"raw_sol_theses": cohort.metrics.get("raw_sol_theses", 0), "raw_sol_theses": cohort.metrics.get("raw_sol_theses", 0),
"onchain_rejected": cohort.metrics.get("onchain_rejected", 0), "onchain_rejected": cohort.metrics.get("onchain_rejected", 0),
"onchain_passed": cohort.metrics.get("onchain_passed", 0),
"token_design_attempts": cohort.metrics.get("token_design_attempts", 0),
"token_design_skipped": cohort.metrics.get("token_design_skipped_due_to_onchain_rejection", 0),
"token_unnecessary": cohort.metrics.get("token_unnecessary_rejections", 0), "token_unnecessary": cohort.metrics.get("token_unnecessary_rejections", 0),
"token_optional_routed_saas": cohort.metrics.get("token_optional_route_to_saas", 0), "token_optional_routed_saas": cohort.metrics.get("token_optional_route_to_saas", 0),
"token_strongly_justified": cohort.metrics.get("token_strongly_justified", 0),
"token_essential": cohort.metrics.get("token_essential", 0),
"crypto_survivors": cohort.members.count(), "crypto_survivors": cohort.members.count(),
"autonomous_crypto_survivors": cohort.metrics.get("autonomous_crypto_survivors", 0), "autonomous_crypto_survivors": cohort.metrics.get("autonomous_crypto_survivors", 0),
"assisted_high_potential": cohort.metrics.get("assisted_high_potential", 0),
"security_blocked": cohort.metrics.get("protocol_security_gate_blocked", 0), "security_blocked": cohort.metrics.get("protocol_security_gate_blocked", 0),
"finalists": cohort.metrics.get("finalists", 0), "finalists": cohort.metrics.get("finalists", 0),
"model_usage": cohort.metrics.get("model_usage", {}), "model_usage": cohort.metrics.get("model_usage", {}),

3
control_plane/ventures/models.py
View file

@ -108,10 +108,13 @@ class CryptoICDecisionType(models.TextChoices):
PROTOCOL_VALIDATE = "PROTOCOL_VALIDATE" PROTOCOL_VALIDATE = "PROTOCOL_VALIDATE"
TESTNET_PILOT = "TESTNET_PILOT" TESTNET_PILOT = "TESTNET_PILOT"
REVISE_TOKEN_MODEL = "REVISE_TOKEN_MODEL" REVISE_TOKEN_MODEL = "REVISE_TOKEN_MODEL"
ASSISTED_CRYPTO_HIGH_POTENTIAL = "ASSISTED_CRYPTO_HIGH_POTENTIAL"
ROUTE_TO_SAAS = "ROUTE_TO_SAAS" ROUTE_TO_SAAS = "ROUTE_TO_SAAS"
ROUTE_TO_ASSISTED_CRYPTO = "ROUTE_TO_ASSISTED_CRYPTO" ROUTE_TO_ASSISTED_CRYPTO = "ROUTE_TO_ASSISTED_CRYPTO"
ROUTE_TO_VENTURE_STUDIO = "ROUTE_TO_VENTURE_STUDIO"
WATCHLIST = "WATCHLIST" WATCHLIST = "WATCHLIST"
REJECT_TOKEN_NOT_NEEDED = "REJECT_TOKEN_NOT_NEEDED" REJECT_TOKEN_NOT_NEEDED = "REJECT_TOKEN_NOT_NEEDED"
REJECT_TOKEN_UTILITY_WEAK = "REJECT_TOKEN_UTILITY_WEAK"
REJECT_ONCHAIN_NOT_NEEDED = "REJECT_ONCHAIN_NOT_NEEDED" REJECT_ONCHAIN_NOT_NEEDED = "REJECT_ONCHAIN_NOT_NEEDED"
REJECT_SECURITY_MODEL = "REJECT_SECURITY_MODEL" REJECT_SECURITY_MODEL = "REJECT_SECURITY_MODEL"
REJECT_SPECULATIVE = "REJECT_SPECULATIVE" REJECT_SPECULATIVE = "REJECT_SPECULATIVE"

2
graph/crypto_venture_cohort.py
View file

@ -31,7 +31,7 @@ def crypto_venture_cohort_graph_v1() -> ExecutionGraphSpec:
"produce_crypto_cohort_report", "produce_crypto_cohort_report",
"complete", "complete",
] ]
spec = ExecutionGraphSpec(name="crypto_venture_cohort", version=3, graph_type="CRYPTO_VENTURE_COHORT", entry="prepare_crypto_mandate", nodes={node: GraphNodeSpec(node, node if node == "complete" else f"crypto_venture_{node}") for node in nodes}, edges=[GraphEdgeSpec(nodes[index], nodes[index + 1], "success") for index in range(len(nodes) - 1)], terminal_nodes=["complete"], metadata={"description": "Crypto / Protocol Venture Cohort V0.3: Sol-native problem-first ideation, explicit onchain gate, independent token counterfactual judge, primary-source-first research."}) spec = ExecutionGraphSpec(name="crypto_venture_cohort", version=4, graph_type="CRYPTO_VENTURE_COHORT", entry="prepare_crypto_mandate", nodes={node: GraphNodeSpec(node, node if node == "complete" else f"crypto_venture_{node}") for node in nodes}, edges=[GraphEdgeSpec(nodes[index], nodes[index + 1], "success") for index in range(len(nodes) - 1)], terminal_nodes=["complete"], metadata={"description": "Crypto / Protocol Venture Cohort V0.3.1: Sol-native problem-first ideation, repaired stage ordering, structural native-token utility judge, primary-source-first research."})
spec.validate() spec.validate()
return spec return spec

62
tests/test_crypto_venture_cohort.py
View file

@ -20,10 +20,12 @@ class SolCryptoReviewProvider(ModelProvider):
payload = {"name": "Machine Identity Credit Union", "one_line_thesis": "Autonomous agents build portable credit histories for machine services.", "product_thesis": "A pre-token API for machine credit and settlement risk scoring.", "user": "AI agent operators", "participants": ["agents", "service providers", "risk observers"], "supply_side": "providers publish measurable service outcomes", "demand_side": "agents need counterparties", "coordination_failure": "portable machine credit is not trusted across operators", "trust_problem": "operators can rewrite private reputation", "why_neutral_network_may_help": "neutral verifiable state lets agents carry economic identity across providers", "pre_token_product": "risk scoring API", "pre_token_customer": "agent operators", "pre_token_value": "lower counterparty default risk", "pre_token_payment_method": "Stripe or stablecoin usage credits", "pre_token_revenue_model": "$100 beta packages", "revenue_1000_path": "10 beta packages", "revenue_5000_path": "20 risk API packages", "revenue_10000_path": "20 subscriptions", "autonomous_operating_loop": "machine demand -> discovery -> settlement -> verification -> reputation update", "bootstrap_plan": "local/testnet simulation", "validation_experiment": "simulate 100 agent jobs", "major_risks": ["token not required"], "confidence": 0.7} payload = {"name": "Machine Identity Credit Union", "one_line_thesis": "Autonomous agents build portable credit histories for machine services.", "product_thesis": "A pre-token API for machine credit and settlement risk scoring.", "user": "AI agent operators", "participants": ["agents", "service providers", "risk observers"], "supply_side": "providers publish measurable service outcomes", "demand_side": "agents need counterparties", "coordination_failure": "portable machine credit is not trusted across operators", "trust_problem": "operators can rewrite private reputation", "why_neutral_network_may_help": "neutral verifiable state lets agents carry economic identity across providers", "pre_token_product": "risk scoring API", "pre_token_customer": "agent operators", "pre_token_value": "lower counterparty default risk", "pre_token_payment_method": "Stripe or stablecoin usage credits", "pre_token_revenue_model": "$100 beta packages", "revenue_1000_path": "10 beta packages", "revenue_5000_path": "20 risk API packages", "revenue_10000_path": "20 subscriptions", "autonomous_operating_loop": "machine demand -> discovery -> settlement -> verification -> reputation update", "bootstrap_plan": "local/testnet simulation", "validation_experiment": "simulate 100 agent jobs", "major_risks": ["token not required"], "confidence": 0.7}
elif "PASS B - ONCHAIN" in prompt: elif "PASS B - ONCHAIN" in prompt:
payload = {"onchain_necessity_score": 86, "why_onchain": "public reputation and escrow materially degrade in private databases", "offchain_substitute": "single operator SaaS", "pass_onchain": True, "reasons": ["verifiable state", "neutral settlement", "machine-to-machine interaction"]} payload = {"onchain_necessity_score": 86, "why_onchain": "public reputation and escrow materially degrade in private databases", "offchain_substitute": "single operator SaaS", "pass_onchain": True, "reasons": ["verifiable state", "neutral settlement", "machine-to-machine interaction"]}
elif "PASS C - DERIVED TOKEN" in prompt: elif "PASS C - MINIMAL DECENTRALIZED" in prompt:
payload = {"protocol_architecture": "onchain escrow and public machine credit registry", "participant_model": "agents, providers, risk observers", "verification_model": "objective settlement proofs", "resource_allocation_model": "risk-weighted access", "dispute_model": "automated evidence checks"}
elif "PASS D - NATIVE TOKEN" in prompt:
payload = {"token_thesis": "Native bond represents protocol-specific machine default risk and is slashed for objective settlement failures.", "token_utility": ["protocol-specific collateral", "machine reputation-backed economic participation"], "token_demand_loop": ["agents earn credit", "providers require bond", "defaults slash bond", "reliable agents receive access"], "value_capture": "risk fees and slashing penalties", "security_model": "escrow, timelock, pause, oracle checks, key controls", "regulatory_risks": ["legal review required"], "required_token_functions": ["protocol-specific risk collateral"], "weak_token_functions": ["payment"]} payload = {"token_thesis": "Native bond represents protocol-specific machine default risk and is slashed for objective settlement failures.", "token_utility": ["protocol-specific collateral", "machine reputation-backed economic participation"], "token_demand_loop": ["agents earn credit", "providers require bond", "defaults slash bond", "reliable agents receive access"], "value_capture": "risk fees and slashing penalties", "security_model": "escrow, timelock, pause, oracle checks, key controls", "regulatory_risks": ["legal review required"], "required_token_functions": ["protocol-specific risk collateral"], "weak_token_functions": ["payment"]}
elif "CRYPTO_TOKEN_COUNTERFACTUAL_JUDGE" in prompt or "Counterfactual token necessity" in prompt: elif "CRYPTO_TOKEN_UTILITY_JUDGE" in prompt or "CRYPTO_TOKEN_COUNTERFACTUAL_JUDGE" in prompt or "Counterfactual token necessity" in prompt:
payload = {"classification": "TOKEN_STRONGLY_JUSTIFIED", "score": 78, "rationale": "Provider staking and slashing materially degrade with database credits.", "fiat_or_database_substitution": "Stablecoin can pay fees but cannot replace bonded slashing and public reputation.", "model_c_external_collateral": {"breaks": "external collateral misses protocol-specific risk"}, "native_token_removed_breaks": ["machine economic identity weakens"], "native_token_removed_degradation": "MATERIAL", "native_token_removed_explanation": "USDC payments work, but protocol-specific bond/reputation does not."} payload = {"token_classification": "TOKEN_STRONGLY_JUSTIFIED", "score": 78, "utility_scores": {"SECURITY_UTILITY": 82, "COLLATERAL_UTILITY": 84, "MACHINE_ECONOMIC_UTILITY": 86, "SPECULATION_DEPENDENCE": 10}, "strong_structural_roles": ["SECURITY_BOND", "SLASHABLE_COLLATERAL"], "weak_roles": ["DEMAND_SIDE_PAYMENT"], "counterfactual_analysis": "USDC/ETH can operate but weakens protocol-specific risk pricing.", "material_improvements_over_USDC_ETH": ["protocol-specific default risk", "portable machine economic identity"], "native_asset_costs_and_risks": ["volatility", "legal complexity"], "argument_for_native_token": "Native bond prices protocol-specific machine risk.", "argument_against_native_token": "ETH/USDC collateral is simpler and more liquid.", "final_rationale": "Benefits outweigh costs for protocol-specific collateral.", "model_c_external_collateral": {"breaks": "external collateral misses protocol-specific risk"}, "native_token_removed_breaks": ["machine economic identity weakens"], "native_token_removed_degradation": "MATERIAL", "native_token_removed_explanation": "USDC payments work, but protocol-specific bond/reputation is materially weaker."}
elif "Independent Token Red Team" in prompt: elif "Independent Token Red Team" in prompt:
payload = {"flags": ["BOOTSTRAP_PROBLEM"], "severity": "MEDIUM", "critique": "Provider supply bootstrap remains the main risk."} payload = {"flags": ["BOOTSTRAP_PROBLEM"], "severity": "MEDIUM", "critique": "Provider supply bootstrap remains the main risk."}
elif "Final Crypto IC" in prompt: elif "Final Crypto IC" in prompt:
@ -62,7 +64,7 @@ def test_crypto_track_and_token_gate_remove_weak_then_regenerate() -> None:
assert weak.status == CompanyProposalStatus.REJECTED assert weak.status == CompanyProposalStatus.REJECTED
assert weak.metadata["routed_to_saas"] is False assert weak.metadata["routed_to_saas"] is False
assert cohort.members.count() == 0 assert cohort.members.count() == 0
assert cohort.metrics["regeneration_policy"].startswith("V0.3") assert cohort.metrics["regeneration_policy"].startswith("V0.3.1")
def test_v03_sol_problem_first_generation_no_token_in_pass_a_and_onchain_gate() -> None: def test_v03_sol_problem_first_generation_no_token_in_pass_a_and_onchain_gate() -> None:
@ -79,6 +81,25 @@ def test_v03_sol_problem_first_generation_no_token_in_pass_a_and_onchain_gate()
assert proposal.metadata["crypto"]["problem_first_pass_a_had_no_token"] is True assert proposal.metadata["crypto"]["problem_first_pass_a_had_no_token"] is True
assert assessment.onchain_necessity_score == 86 assert assessment.onchain_necessity_score == 86
assert cohort.metrics["raw_sol_theses"] == 1 assert cohort.metrics["raw_sol_theses"] == 1
assert cohort.metrics["sol_onchain_judge_requests"] == 1
def test_v031_onchain_reject_skips_token_design() -> None:
class RejectOnchainProvider(SolCryptoReviewProvider):
def complete(self, request: ModelRequestContract) -> ModelResponseContract:
if "PASS B - ONCHAIN" in request.prompt:
return ModelResponseContract("sol", json.dumps({"onchain_necessity_score": 40, "pass_onchain": False, "why_onchain": "database is fine"}), {})
return super().complete(request)
svc = sol_service(ModelRouter({"sol": RejectOnchainProvider()}))
cohort = svc.prepare_cohort(size=1, concurrency=1)
svc.generate_protocols(cohort)
svc.onchain_necessity_gate(cohort)
svc.token_necessity_gate(cohort)
assert cohort.metrics["onchain_passed"] == 0
assert cohort.metrics["token_design_attempts"] == 0
assert cohort.metrics["token_design_skipped_due_to_onchain_rejection"] == 1
def test_sol_counterfactual_red_team_and_final_ic_are_used() -> None: def test_sol_counterfactual_red_team_and_final_ic_are_used() -> None:
@ -93,7 +114,9 @@ def test_sol_counterfactual_red_team_and_final_ic_are_used() -> None:
review = svc.portfolio_crypto_ic(cohort) review = svc.portfolio_crypto_ic(cohort)
assert token.classification == TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED assert token.classification == TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED
assert token.metadata["sol_counterfactual_review"] assert token.metadata["sol_token_utility_judge"]
assert token.metadata["argument_for_native_token"]
assert token.metadata["argument_against_native_token"]
assert TokenRedTeamFlag.BOOTSTRAP_PROBLEM in red.flags assert TokenRedTeamFlag.BOOTSTRAP_PROBLEM in red.flags
assert review.rankings[0]["sol_final_ic_review"]["score_adjustment"] == 3 assert review.rankings[0]["sol_final_ic_review"]["score_adjustment"] == 3
@ -119,10 +142,11 @@ def test_tokenomics_simulation_is_proposal_specific() -> None:
low_sim = svc.simulate_tokenomics(low) low_sim = svc.simulate_tokenomics(low)
high_sim = svc.simulate_tokenomics(high) high_sim = svc.simulate_tokenomics(high)
assert low_sim.scenarios["EXPECTED"]["users"] != high_sim.scenarios["EXPECTED"]["users"] assert low_sim.scenarios["EXPECTED_USAGE"]["users"] != high_sim.scenarios["EXPECTED_USAGE"]["users"]
assert low_sim.scenarios["EXPECTED"]["token_demand"] != high_sim.scenarios["EXPECTED"]["token_demand"] assert low_sim.scenarios["EXPECTED_USAGE"]["token_demand"] != high_sim.scenarios["EXPECTED_USAGE"]["token_demand"]
assert "NATIVE_TOKEN_REMOVED" in low_sim.scenarios assert "NATIVE_TOKEN_REMOVED" in low_sim.scenarios
assert "EXTERNAL_COLLATERAL" in low_sim.scenarios assert "ETH_COLLATERAL" in low_sim.scenarios
assert "USDC_COLLATERAL" in low_sim.scenarios
def test_v03_primary_source_quality_wrong_category_and_weak_coverage() -> None: def test_v03_primary_source_quality_wrong_category_and_weak_coverage() -> None:
@ -187,9 +211,23 @@ def test_v03_independent_counterfactual_native_removed_fields() -> None:
token = svc.assess_token_necessity(proposal) token = svc.assess_token_necessity(proposal)
assert token.metadata["independent_counterfactual_judge"] == "CRYPTO_TOKEN_COUNTERFACTUAL_JUDGE" assert token.metadata["independent_token_utility_judge"] == "CRYPTO_TOKEN_UTILITY_JUDGE"
assert token.metadata["native_token_removed_degradation"] == "MATERIAL" assert token.metadata["native_token_removed_degradation"] == "MATERIAL"
assert token.metadata["model_c_external_collateral"] assert token.metadata["argument_for_native_token"]
def test_v031_payment_governance_reward_only_rejected_but_structural_passes() -> None:
svc = service(ModelRouter({"sol": SolCryptoReviewProvider()}))
cohort = svc.prepare_cohort(size=1, concurrency=1)
weak = svc._create_proposal(cohort.mandate, {"name": "FeeGov Rewards", "product_thesis": "Useful SaaS", "protocol_thesis": "Onchain logs", "token_thesis": "Token is for payment governance and rewards", "token_utility": ["payment", "governance", "rewards"], "token_roles": {"DEMAND_SIDE_PAYMENT": "REQUIRED", "GOVERNANCE": "REQUIRED", "PROVIDER_REWARD": "REQUIRED"}}, "test", "OPEN_CRYPTO_CATEGORY")
strong = svc._create_proposal(cohort.mandate, {**svc._fallback_payload(0, "MACHINE_REPUTATION_WITH_ECONOMIC_STAKE"), "token_roles": {"SECURITY_BOND": "REQUIRED", "SLASHABLE_COLLATERAL": "REQUIRED", "MACHINE_ECONOMIC_IDENTITY": "STRONGLY_USEFUL"}}, "test", "MACHINE_REPUTATION_WITH_ECONOMIC_STAKE")
weak_token = svc.assess_token_necessity(weak)
strong_token = svc.assess_token_necessity(strong)
assert weak_token.classification in {TokenNecessityClassification.TOKEN_OPTIONAL, TokenNecessityClassification.TOKEN_UNNECESSARY}
assert strong_token.classification == TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED
assert len(strong_token.metadata["strong_token_roles"]) >= 2
def test_crypto_cohort_graph_real_gates_and_report() -> None: def test_crypto_cohort_graph_real_gates_and_report() -> None:
@ -209,4 +247,6 @@ def test_crypto_cohort_graph_real_gates_and_report() -> None:
assert cohort.metrics["protocol_security_gate_passed"] >= 1 assert cohort.metrics["protocol_security_gate_passed"] >= 1
assert len(report.content["ranking"]) <= 3 assert len(report.content["ranking"]) <= 3
assert len(report.content["top_3"]) <= 3 assert len(report.content["top_3"]) <= 3
assert report.content["title"] == "CRYPTO VENTURE COHORT V0.3" assert report.content["title"] == "CRYPTO VENTURE COHORT V0.3.1"
if cohort.metrics.get("serious_crypto_survivors", 0) > 0:
assert cohort.metrics["sol_final_ic_requests"] > 0