From 3cdd7b0cae35b012302a5e12999d5212c497c2df Mon Sep 17 00:00:00 2001 From: Daniel Maddern Date: Sun, 16 Aug 2026 19:31:15 +0700 Subject: [PATCH] Add crypto protocol cohort V0.2 --- agents/crypto_venture.py | 353 +++++++++++++++--- .../commands/export_crypto_venture_cohort.py | 20 +- .../commands/run_crypto_venture_cohort.py | 9 +- ...pretokenmonetizationassessment_and_more.py | 130 +++++++ control_plane/ventures/models.py | 128 +++++++ graph/crypto_venture_cohort.py | 2 +- tests/test_crypto_venture_cohort.py | 30 +- 7 files changed, 601 insertions(+), 71 deletions(-) create mode 100644 control_plane/ventures/migrations/0006_pretokenmonetizationassessment_and_more.py diff --git a/agents/crypto_venture.py b/agents/crypto_venture.py index 7f2568b..2764b0b 100644 --- a/agents/crypto_venture.py +++ b/agents/crypto_venture.py @@ -21,15 +21,25 @@ from control_plane.ventures.models import ( CryptoJurisdictionPolicy, CryptoRedTeamAssessment, CryptoScenarioLab, + CryptoAutonomyClass, FounderDependencyLevel, OnchainNecessityAssessment, PortfolioICReview, PortfolioThesis, PortfolioThesisStatus, ProtocolThesis, + ProtocolSecurityAssessment, + ProtocolSecurityDecision, ProtocolValueCapture, + PreTokenMonetizationAssessment, + PreTokenMonetizationMode, TokenDemandLoop, + TokenEconomicModel, + TokenLaunchReadinessAssessment, + TokenLaunchReadinessStatus, TokenNecessityClassification, + TokenRoleDecomposition, + TokenRoleRequirement, TokenomicsSimulation, TokenRedTeamFlag, TokenUtilityAssessment, @@ -44,6 +54,7 @@ from control_plane.ventures.models import ( from model_router.policy import model_for_role from model_router.providers import extract_json_object from model_router.router import ModelCapability, ModelRequestContract, ModelRouter +from research.searxng import SearxngSearchClient, WebPageFetcher CRYPTO_TERRITORIES = [ @@ -94,8 +105,11 @@ CRYPTO_SCORE_DIMENSIONS = [ "AUTONOMOUS_OPERABILITY", "SECURITY_MODEL_QUALITY", "REGULATORY_MANAGEABILITY", + "PRE_TOKEN_MONETIZATION_POTENTIAL", ] +CRYPTO_RESEARCH_CATEGORIES = ["USER_PAIN", "EXISTING_PROTOCOLS", "FAILED_PRECEDENTS", "PAYMENT_MODELS", "TOKEN_MODELS", "STAKING_MODELS", "SLASHING_PRECEDENTS", "PROVIDER_ECONOMICS", "NETWORK_BOOTSTRAP", "ONCHAIN_ALTERNATIVES", "OFFCHAIN_ALTERNATIVES", "SECURITY_INCIDENTS", "LEGAL_REGULATORY", "TOKEN_LAUNCH_PRECEDENTS", "PRE_TOKEN_MONETIZATION_PRECEDENTS"] + CRYPTO_SCENARIOS = [ "oracle failure", "validator/provider collusion", @@ -128,15 +142,15 @@ class CryptoVentureService: objective="Generate crypto-native protocol ventures where onchain execution and a native token are genuinely necessary for real usage, not speculation.", constraints={"venture_track": VentureTrack.CRYPTO_PROTOCOL, "no_token_sale": True, "no_fundraising": True, "no_mainnet_issuance": True, "no_us_targeted_activity": True, "testnet_or_local_only": True, "no_real_spend": True, "no_user_contact": True}, optimization_targets=["token necessity", "onchain necessity", "real usage demand", "value accrual", "sustainable tokenomics", "autonomous operability", "regulatory manageability", "security model quality"], - metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.1", "venture_track": VentureTrack.CRYPTO_PROTOCOL}, + metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.2", "venture_track": VentureTrack.CRYPTO_PROTOCOL}, ) self._artifact(None, mandate, "CRYPTO_VENTURE_MANDATE", "Crypto Protocol Venture V0.1 Mandate", mandate.constraints, "Crypto protocol cohort mandate. No token sale, fundraising, or mainnet issuance.", "crypto_venture") return mandate def prepare_cohort(self, *, size: int, graph_run=None, concurrency: int = 2) -> VentureCohort: mandate = self.create_crypto_mandate() - cohort_id = f"CPV01-{time.strftime('%Y%m%d%H%M%S')}-{uuid.uuid4().hex[:8]}" - return VentureCohort.objects.create(cohort_id=cohort_id, mandate=mandate, cohort_size=size, graph_run=graph_run, concurrency=concurrency, status="PREPARING_CRYPTO", graph_versions={"cohort": "crypto_venture_cohort v1"}, research_policy={"source_linked_evidence_required": True, "research_insufficient_if_no_sources": True}, scoring_policy={"dimensions": CRYPTO_SCORE_DIMENSIONS}, metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.1", "venture_track": VentureTrack.CRYPTO_PROTOCOL, "real_spend": 0, "real_customer_outreach": False, "token_sale": False, "fundraising": False, "mainnet_issuance": False}) + cohort_id = f"CPV02-{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 v2"}, research_policy={"source_linked_evidence_required": True, "research_insufficient_if_no_sources": True, "crypto_categories": CRYPTO_RESEARCH_CATEGORIES}, scoring_policy={"dimensions": CRYPTO_SCORE_DIMENSIONS, "strict_token_filter": True}, metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.2", "venture_track": VentureTrack.CRYPTO_PROTOCOL, "real_spend": 0, "real_customer_outreach": False, "token_sale": False, "fundraising": False, "mainnet_issuance": False}) def generate_protocols(self, cohort: VentureCohort) -> list[CompanyProposal]: accepted = [] @@ -148,7 +162,7 @@ class CryptoVentureService: proposal = self._create_proposal(cohort.mandate, payload, source, territory) VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"territory": territory}) accepted.append(proposal) - cohort.metrics = {**cohort.metrics, "requested_protocol_count": cohort.cohort_size, "generation_attempts": attempts, "accepted_protocols": len(accepted), "duplicate_rejections": 0, "token_necessity_rejections": 0, "generation_sources": sorted({p.metadata.get("generation_source", "unknown") for p in accepted})} + cohort.metrics = {**cohort.metrics, "raw_target": cohort.cohort_size, "raw_generated": len(accepted), "requested_protocol_count": cohort.cohort_size, "generation_attempts": attempts, "accepted_protocols": len(accepted), "duplicate_rejections": 0, "token_necessity_rejections": 0, "generation_sources": sorted({p.metadata.get("generation_source", "unknown") for p in accepted})} cohort.status = "PROTOCOLS_GENERATED" cohort.save(update_fields=["metrics", "status", "updated_at"]) return accepted @@ -186,24 +200,8 @@ class CryptoVentureService: return {"removed": removed, "hard_exclusion_rejections": hard_excluded, "duplicate_rejections": duplicate} def regenerate_rejected_slots(self, cohort: VentureCohort, *, reason: str = "novelty_or_token_gate") -> list[CompanyProposal]: - created = [] - attempts = 0 - max_attempts = min(6, max(2, (cohort.cohort_size - cohort.members.count()) * 2)) - while cohort.members.count() < cohort.cohort_size and attempts < max_attempts: - attempts += 1 - index = cohort.members.count() + attempts - territory = CRYPTO_TERRITORIES[index % len(CRYPTO_TERRITORIES)] - if attempts <= 2: - payload, source = self._protocol_payload(index, territory) - else: - payload, source = self._fallback_payload(index, territory), "deterministic_crypto_repair" - proposal = self._create_proposal(cohort.mandate, payload, source, territory) - VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"territory": territory, "regenerated_for": reason}) - created.append(proposal) - self.assess_token_necessity(proposal) - self.novelty_gate(cohort) - self._remove_weak_token_members(cohort) - cohort.metrics = {**cohort.metrics, "regeneration_attempts": cohort.metrics.get("regeneration_attempts", 0) + attempts, "regenerated_protocols": cohort.metrics.get("regenerated_protocols", 0) + len(created), "accepted_protocols": cohort.members.count(), "unfilled_slots_after_regeneration": max(0, cohort.cohort_size - cohort.members.count())} + 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.2 uses total raw generation budget; weak token ideas are not force-filled."} cohort.save(update_fields=["metrics", "updated_at"]) return created @@ -230,24 +228,135 @@ class CryptoVentureService: proposal.status = CompanyProposalStatus.REJECTED proposal.metadata = {**proposal.metadata, "routed_to_saas": assessment.classification == TokenNecessityClassification.TOKEN_OPTIONAL} proposal.save(update_fields=["status", "metadata", "updated_at"]) - return {"token_unnecessary_rejections": cohort.metrics.get("token_unnecessary_rejections", 0) + unnecessary, "token_optional_route_to_saas": cohort.metrics.get("token_optional_route_to_saas", 0) + optional} + 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()} def light_market_research(self, cohort: VentureCohort) -> None: started = time.monotonic() total_sources = 0 + total_rejected = 0 insufficient = 0 for member in cohort.members.select_related("proposal"): - research = self.venture.conduct_market_research(member.proposal, depth="light") + research = self.crypto_research(member.proposal, depth="light") total_sources += len(research.get("sources", [])) + total_rejected += len(research.get("source_rejections", [])) if not research.get("sources"): insufficient += 1 - cohort.metrics = {**cohort.metrics, "crypto_light_research_seconds": round(time.monotonic() - started, 2), "crypto_light_research_sources": total_sources, "research_insufficient_count": insufficient} + cohort.metrics = {**cohort.metrics, "crypto_light_research_seconds": round(time.monotonic() - started, 2), "crypto_light_research_sources": total_sources, "crypto_light_research_rejected_sources": total_rejected, "research_insufficient_count": insufficient, "research_survivors": cohort.members.count() - insufficient} cohort.save(update_fields=["metrics", "updated_at"]) def protocol_research(self, cohort: VentureCohort) -> None: for member in cohort.members.select_related("proposal"): self._update_protocol_research_flags(member.proposal) + def crypto_research(self, proposal: CompanyProposal, *, depth: str = "light") -> dict[str, Any]: + query_plan = self.crypto_query_plan(proposal, depth=depth) + client = SearxngSearchClient.from_resources() + fetcher = WebPageFetcher() + all_sources = [] + query_results = [] + engine_health = Counter() + if client is not None: + per_query_limit = 4 if depth == "deep" else 2 + for category, queries in query_plan.items(): + for query in queries: + try: + payload = client.search_payload(query) + except Exception as exc: + query_results.append({"category": category, "query": query, "status": "SEARCH_INFRASTRUCTURE_FAILED", "error": str(exc), "result_count": 0}) + continue + unresponsive = payload.get("unresponsive_engines", []) if isinstance(payload, dict) else [] + for engine, reason in unresponsive: + engine_health[self._engine_status(str(reason))] += 1 + raw_results = payload.get("results", []) if isinstance(payload, dict) else [] + query_results.append({"category": category, "query": query, "status": "SEARCH_WORKED", "result_count": len(raw_results), "unresponsive_engines": unresponsive}) + for item in raw_results[:per_query_limit]: + if isinstance(item, dict) and item.get("url"): + all_sources.append({"type": "public_web", "source": "searxng", "url": str(item["url"]), "title": str(item.get("title", "")), "category": category, "summary": str(item.get("content", item.get("snippet", ""))), "fallback_evidence": False}) + pages = fetcher.fetch_many(all_sources, max_pages=10 if depth == "deep" else 5) if all_sources else [] + page_by_url = {page["url"]: page for page in pages} + accepted = [] + rejected = [] + seen = set() + for source in all_sources: + url = source.get("url") + if not url or url in seen: + continue + seen.add(url) + enriched = {**source, **({"content": page_by_url[url].get("text", "")} if url in page_by_url else {})} + quality = self.crypto_source_quality(proposal, enriched) + if quality["accepted"]: + accepted.append({**enriched, "quality": quality["quality"], "relevance_score": quality["score"], "acceptance_reason": quality["reason"]}) + else: + rejected.append({**enriched, "quality": "weak", "rejection_reason": quality["reason"], "relevance_score": quality["score"]}) + covered = {source["category"] for source in accepted if source.get("quality") == "strong"} + coverage = {category: category in covered for category in CRYPTO_RESEARCH_CATEGORIES} + coverage_ratio = round(sum(1 for value in coverage.values() if value) / len(CRYPTO_RESEARCH_CATEGORIES), 2) + acceptance_rate = round(len(accepted) / max(1, len(accepted) + len(rejected)), 2) + search_infrastructure_failed = client is None or (not all_sources and any(item.get("status") == "SEARCH_INFRASTRUCTURE_FAILED" for item in query_results)) + evidence_insufficient = len(accepted) == 0 or coverage_ratio < (0.7 if depth == "deep" else 0.25) + research = {"query_plan": query_plan, "query_results": query_results, "sources": accepted, "source_rejections": rejected, "source_count": len(accepted), "source_rejection_count": len(rejected), "page_fetch_count": len(pages), "search_result_count": len(all_sources), "coverage": coverage, "coverage_ratio": coverage_ratio, "category_coverage_confidence": {category: "HIGH" if coverage[category] else "LOW" for category in CRYPTO_RESEARCH_CATEGORIES}, "source_provider_health": "SEARCH_INFRASTRUCTURE_FAILED" if search_infrastructure_failed else "SEARCH_WORKED", "search_engine_health": dict(engine_health), "source_acceptance_rate": acceptance_rate, "research_status": "SEARCH_INFRASTRUCTURE_FAILED" if search_infrastructure_failed else "EVIDENCE_INSUFFICIENT" if evidence_insufficient else "SOURCE_LINKED", "unverified_categories": [category for category, ok in coverage.items() if not ok]} + proposal.market_evidence = [*self._as_list(proposal.market_evidence), *accepted, {"type": "crypto_research_coverage", "source": "crypto_research", "summary": f"{depth} crypto research coverage {coverage_ratio}", "coverage": coverage, "depth": depth}] + proposal.metadata = {**proposal.metadata, "research": research} + proposal.save(update_fields=["market_evidence", "metadata", "updated_at"]) + self._artifact(proposal, proposal.mandate, "CRYPTO_MARKET_RESEARCH", f"Crypto Research ({depth})", research, f"Crypto research {depth}: {len(accepted)} accepted, {len(rejected)} rejected, coverage {coverage_ratio}.", "crypto_research") + return research + + def crypto_query_plan(self, proposal: CompanyProposal, *, depth: str) -> dict[str, list[str]]: + crypto = proposal.metadata.get("crypto", {}) + seed = " ".join([proposal.title, str(crypto.get("territory", "")), proposal.protocol_thesis.protocol_category]).replace("_", " ") + terms = { + "USER_PAIN": [f"{seed} user pain", f"{proposal.target_customer} decentralized protocol pain"], + "EXISTING_PROTOCOLS": [f"{seed} protocol", f"{seed} competitors crypto"], + "FAILED_PRECEDENTS": [f"{seed} failed crypto project", f"{seed} postmortem"], + "PAYMENT_MODELS": [f"{seed} stablecoin payments", f"{seed} usage fees"], + "TOKEN_MODELS": [f"{seed} token model staking", f"{seed} tokenomics"], + "STAKING_MODELS": [f"{seed} staking slashing", f"provider staking protocol slashing"], + "SLASHING_PRECEDENTS": [f"{seed} slashing precedent", f"objective slashing crypto protocol"], + "PROVIDER_ECONOMICS": [f"{seed} provider economics", f"decentralized provider marketplace economics"], + "NETWORK_BOOTSTRAP": [f"{seed} network bootstrap", f"crypto protocol bootstrap supply demand"], + "ONCHAIN_ALTERNATIVES": [f"{seed} onchain alternatives"], + "OFFCHAIN_ALTERNATIVES": [f"{seed} SaaS alternative", f"{seed} offchain alternative"], + "SECURITY_INCIDENTS": [f"{seed} security incident", f"{seed} exploit"], + "LEGAL_REGULATORY": [f"{seed} token regulatory risk", f"crypto staking slashing regulatory"], + "TOKEN_LAUNCH_PRECEDENTS": [f"{seed} token launch", f"protocol token launch precedent"], + "PRE_TOKEN_MONETIZATION_PRECEDENTS": [f"{seed} paid beta", f"crypto protocol pre token revenue"], + } + limit = 4 if depth == "deep" else 2 + return {category: queries[:limit] for category, queries in terms.items()} + + def crypto_source_quality(self, proposal: CompanyProposal, source: dict[str, Any]) -> dict[str, Any]: + text = " ".join(str(source.get(key, "")) for key in ["title", "summary", "content", "url"]).lower() + url = str(source.get("url", "")).lower() + if any(noisy in url for noisy in ["pinterest", "facebook.com", "instagram.com", "x.com/intent", "webcache", "archive.org", "youtube.com/watch"]): + return {"accepted": False, "score": 0.0, "reason": "noisy or low-evidence source"} + preferred = ["docs", "github", "whitepaper", "paper", "postmortem", "audit", "security", "research", "protocol", "token", "staking", "slashing", "stablecoin", "attestation", "oracle"] + proposal_terms = set(re.findall(r"[a-z0-9]{5,}", self._proposal_text(proposal))) + source_terms = set(re.findall(r"[a-z0-9]{5,}", text)) + overlap = len(proposal_terms & source_terms) + preferred_hits = sum(1 for term in preferred if term in text or term in url) + score = round(min(1.0, overlap * 0.03 + preferred_hits * 0.08), 2) + if score >= 0.22: + return {"accepted": True, "quality": "strong", "score": score, "reason": "crypto-relevant technical/protocol evidence"} + if score >= 0.12: + return {"accepted": True, "quality": "weak", "score": score, "reason": "accepted as weak crypto context"} + return {"accepted": False, "score": score, "reason": "insufficient crypto/protocol relevance"} + + def _engine_status(self, reason: str) -> str: + lowered = reason.lower() + if "captcha" in lowered: + return "CAPTCHA" + if "rate" in lowered or "too many" in lowered: + return "RATE_LIMITED" + if "denied" in lowered: + return "ACCESS_DENIED" + if "timeout" in lowered: + return "TIMEOUT" + if "protocol" in lowered: + return "PROTOCOL_ERROR" + if "disabled" in lowered: + return "DISABLED" + return "UNAVAILABLE" + def token_red_team(self, cohort: VentureCohort) -> list[CryptoRedTeamAssessment]: assessments = [self.red_team_proposal(member.proposal) for member in cohort.members.select_related("proposal")] counts = Counter(flag for item in assessments for flag in item.flags) @@ -256,7 +365,13 @@ class CryptoVentureService: return assessments def tokenomics_simulation(self, cohort: VentureCohort) -> list[TokenomicsSimulation]: - simulations = [self.simulate_tokenomics(member.proposal) for member in cohort.members.select_related("proposal")] + simulations = [] + for member in cohort.members.select_related("proposal"): + self.assess_pretoken_monetization(member.proposal) + self.token_role_decomposition(member.proposal) + self.token_economic_model(member.proposal) + simulations.append(self.simulate_tokenomics(member.proposal)) + self.token_launch_readiness(member.proposal) self.crypto_scenario_lab(cohort) return simulations @@ -273,7 +388,7 @@ class CryptoVentureService: members = list(cohort.members.select_related("proposal").order_by("-portfolio_score", "created_at")[:5]) for member in members: before[str(member.proposal_id)] = member.proposal.metadata.get("research", {}).get("coverage_ratio", 0.0) - self.venture.conduct_market_research(member.proposal, depth="deep") + self.crypto_research(member.proposal, depth="deep") after[str(member.proposal_id)] = member.proposal.metadata.get("research", {}).get("coverage_ratio", 0.0) cohort.metrics = {**cohort.metrics, "top5_deep_research_count": len(members), "top5_research_coverage_before": before, "top5_research_coverage_after": after} cohort.save(update_fields=["metrics", "updated_at"]) @@ -292,7 +407,7 @@ class CryptoVentureService: member.save(update_fields=["rank", "portfolio_score", "is_top_3", "updated_at"]) row["rank"] = rank review, _ = PortfolioICReview.objects.update_or_create(cohort=cohort, defaults={"rankings": rows, "top_3": top_3, "concentration": self._crypto_concentration(rows), "metadata": {"finalist_shortfall": max(0, 3 - len(top_3)), "no_fund_decision_v01": True}}) - cohort.metrics = {**cohort.metrics, "crypto_ranked_count": len(rows), "crypto_top_3_count": len(top_3)} + cohort.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)} cohort.status = "CRYPTO_IC_COMPLETE" cohort.save(update_fields=["metrics", "status", "updated_at"]) return review @@ -306,18 +421,31 @@ class CryptoVentureService: required = ["contract", "oracle", "key", "admin", "pause", "treasury"] missing = [item for item in required if item not in text] critical = any(self._positive_phrase_present(text, term) for term in ["unaudited mainnet", "custody user funds", "bridge dependency", "upgradeable without timelock"]) - status = "BLOCKED_HUMAN_SECURITY_GATE" if critical or len(missing) >= 4 else "PASS_WITH_SECURITY_REVIEW" - if status.startswith("BLOCKED"): + decision = ProtocolSecurityDecision.BLOCK_TESTNET if critical or len(missing) >= 5 else ProtocolSecurityDecision.REVISE_SECURITY_MODEL if len(missing) >= 3 else ProtocolSecurityDecision.PASS_FOR_TESTNET_DESIGN + if decision == ProtocolSecurityDecision.BLOCK_TESTNET: blocked += 1 else: passed += 1 - self._artifact(proposal, cohort.mandate, "CRYPTO_PROTOCOL_SECURITY_GATE", f"Protocol Security Gate: {proposal.title}", {"status": status, "missing_controls": missing, "critical_risk": critical, "mainnet_allowed": False}, f"{status}. Missing controls: {', '.join(missing) or 'none'}. Mainnet is not allowed in V0.1.", "crypto_protocol_security_gate", graph_run=cohort.graph_run) - proposal.metadata = {**proposal.metadata, "crypto_security_gate": {"status": status, "missing_controls": missing, "critical_risk": critical, "mainnet_allowed": False}} + assessment = ProtocolSecurityAssessment.objects.update_or_create(proposal=proposal, defaults={"decision": decision, "threat_model": {"contract_architecture": "design-stage only", "staking_slashing_logic": "requires objective measurable failures", "oracle_assumptions": "must avoid circular/manual oracle dependency", "upgradeability": "timelock/admin controls required", "pause_emergency_controls": "required before testnet", "mainnet_allowed": False}, "missing_controls": missing, "privileged_roles": ["admin", "treasury", "pauser"], "economic_attack_surfaces": ["sybil providers", "collusion", "oracle manipulation", "slash griefing"], "key_management_assumptions": ["no production key custody in V0.2", "multisig/timelock required later"], "rationale": "Guard-style protocol threat model review; no Solidity audit or mainnet approval."})[0] + self._artifact(proposal, cohort.mandate, "CRYPTO_PROTOCOL_SECURITY_GATE", f"Protocol Security Gate: {proposal.title}", {"decision": assessment.decision, "missing_controls": missing, "critical_risk": critical, "mainnet_allowed": False}, f"{assessment.decision}. Missing controls: {', '.join(missing) or 'none'}. Mainnet is not allowed in V0.2.", "crypto_guard_workflow", graph_run=cohort.graph_run) + proposal.metadata = {**proposal.metadata, "crypto_security_gate": {"decision": assessment.decision, "missing_controls": missing, "critical_risk": critical, "mainnet_allowed": False}} proposal.save(update_fields=["metadata", "updated_at"]) cohort.metrics = {**cohort.metrics, "protocol_security_gate_blocked": blocked, "protocol_security_gate_passed": passed} cohort.save(update_fields=["metrics", "updated_at"]) return {"blocked": blocked, "passed": passed} + def token_launch_readiness(self, proposal: CompanyProposal) -> TokenLaunchReadinessAssessment: + simulation = getattr(proposal, "tokenomics_simulation", None) + token = getattr(proposal, "token_utility_assessment", None) or self.assess_token_necessity(proposal) + security = getattr(proposal, "protocol_security_assessment", None) + if token.classification in {TokenNecessityClassification.TOKEN_OPTIONAL, TokenNecessityClassification.TOKEN_UNNECESSARY}: + status = TokenLaunchReadinessStatus.TOKEN_MODEL_REVISION_REQUIRED + elif security and security.decision == ProtocolSecurityDecision.BLOCK_TESTNET: + status = TokenLaunchReadinessStatus.SECURITY_REVIEW_REQUIRED + else: + status = TokenLaunchReadinessStatus.PRODUCT_TRACTION_REQUIRED + return TokenLaunchReadinessAssessment.objects.update_or_create(proposal=proposal, defaults={"real_users": False, "repeat_usage": False, "real_protocol_fees": False, "token_necessity_validated": False, "token_demand_loop_validated": False, "security_audit_status": "NOT_AUDITED_DESIGN_ONLY", "testnet_stability": "NOT_PROVEN", "tokenomics_stress_tests": simulation.summary if simulation else {}, "legal_review_status": "LEGAL_REVIEW_REQUIRED", "jurisdiction_policy_status": "US_EXCLUDED_TOKEN_SALE_DISABLED", "admin_key_controls": "DESIGN_REQUIRED", "treasury_controls": "DESIGN_REQUIRED", "decentralization_readiness": "NOT_READY", "launch_readiness_status": status})[0] + def regulatory_gate(self, cohort: VentureCohort) -> None: gated = 0 for member in cohort.members.select_related("proposal"): @@ -328,8 +456,20 @@ class CryptoVentureService: cohort.save(update_fields=["metrics", "updated_at"]) def capability_analysis(self, cohort: VentureCohort) -> list[dict[str, Any]]: - gaps = ["smart contract audit", "testnet deployment", "wallet auth", "key management", "oracle/provider monitoring", "token simulation harness", "legal review workflow"] - rows = [{"capability": gap, "count": cohort.members.count(), "status": "MISSING", "earliest_stage": "BEFORE_VALIDATION"} for gap in gaps] + capabilities = [ + ("Guard model", "AVAILABLE"), + ("protocol threat modelling", "AVAILABLE"), + ("Solidity implementation", "AVAILABLE"), + ("token simulation harness", "AVAILABLE"), + ("crypto research workflow", "PARTIAL"), + ("testnet deployment", "PARTIAL"), + ("wallet auth", "MISSING"), + ("key management", "MISSING"), + ("contract deployment pipeline", "MISSING"), + ("oracle/provider monitoring", "MISSING"), + ("legal review workflow", "MISSING"), + ] + rows = [{"capability": capability, "count": cohort.members.count(), "status": status, "earliest_stage": "BEFORE_VALIDATION"} for capability, status in capabilities] review = getattr(cohort, "portfolio_review", None) if review: review.capability_demand = rows @@ -356,19 +496,36 @@ class CryptoVentureService: def produce_crypto_cohort_report(self, cohort: VentureCohort) -> VentureArtifact: review = cohort.portfolio_review rows = review.rankings - content = {"title": "CRYPTO VENTURE COHORT V0.1 REPORT", "cohort_id": cohort.cohort_id, "graph_run": str(cohort.graph_run_id or ""), "runtime": cohort.metrics, "accepted_protocols": cohort.members.count(), "generation_attempts": cohort.metrics.get("generation_attempts", 0), "duplicate_token_necessity_rejections": {"duplicate_rejections": cohort.metrics.get("duplicate_rejections", 0), "token_unnecessary_rejections": cohort.metrics.get("token_unnecessary_rejections", 0)}, "ranking": rows, "top_3": review.top_3, "token_red_team_failures": cohort.metrics.get("token_red_team_failures", {}), "token_utility_distribution": dict(Counter(util for row in rows for util in row.get("token_utility", []))), "crypto_thesis_saturation": review.concentration, "capability_gaps": review.capability_demand, "system_metrics": cohort.metrics, "stop_conditions": {"token_sale": False, "fundraising": False, "mainnet_issuance": False, "user_contact": False, "real_spend": 0}} - artifact = self._artifact(None, cohort.mandate, "CRYPTO_VENTURE_COHORT_REPORT", "Crypto Venture Cohort V0.1 Report", content, self._readable_report(content), "crypto_portfolio_ic", graph_run=cohort.graph_run) + research_health = self._research_health(rows) + content = {"title": "CRYPTO VENTURE COHORT V0.2", "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), "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), "ranking": rows, "top_3": review.top_3, "token_red_team_failures": cohort.metrics.get("token_red_team_failures", {}), "token_utility_distribution": dict(Counter(util for row in rows for util in row.get("token_utility", []))), "crypto_thesis_saturation": review.concentration, "capability_gaps": review.capability_demand, "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.2 Report", content, self._readable_report(content), "crypto_portfolio_ic", graph_run=cohort.graph_run) cohort.status = "COMPLETE" cohort.save(update_fields=["status", "updated_at"]) return artifact + def _research_health(self, rows: list[dict[str, Any]]) -> dict[str, Any]: + engine_counts = Counter() + accepted = 0 + rejected = 0 + coverage = [] + for row in rows: + research = row.get("research", {}) + accepted += int(research.get("source_count", 0) or 0) + rejected += int(research.get("source_rejection_count", 0) or 0) + coverage.append(float(research.get("coverage_ratio", 0) or 0)) + engine_counts.update(research.get("search_engine_health", {})) + return {"search_engines": dict(engine_counts), "sources_accepted": accepted, "sources_rejected": rejected, "average_coverage": round(sum(coverage) / max(1, len(coverage)), 2), "healthy": engine_counts.get("AVAILABLE", 0), "rate_limited": engine_counts.get("RATE_LIMITED", 0), "captcha": engine_counts.get("CAPTCHA", 0), "denied": engine_counts.get("ACCESS_DENIED", 0), "timed_out": engine_counts.get("TIMEOUT", 0)} + def assess_token_necessity(self, proposal: CompanyProposal) -> TokenUtilityAssessment: crypto = proposal.metadata.get("crypto", {}) text = self._proposal_text(proposal) utilities = [u for u in crypto.get("token_utility", []) if isinstance(u, str)] or self._infer_utilities(text) + roles = self.token_role_decomposition(proposal) strong_count = sum(1 for utility in utilities if utility.lower() in ALLOWED_TOKEN_UTILITIES or any(term in utility.lower() for term in ["stake", "slash", "collateral", "fee", "marketplace", "attestation", "compute", "machine", "security"])) - weak = any(term in text for term in ["meme", "speculative", "community token", "governance token only", "token gated subscription"]) - score = min(100, 35 + strong_count * 18 + (10 if "slash" in text or "slashing" in text else 0) + (8 if "provider" in text else 0) - (35 if weak else 0)) + required_roles = len(roles.required_roles) + 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"] + 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)) sol_review = self._sol_json("Counterfactual token necessity review. If the token were removed and replaced with fiat/stablecoin/database credits, would the product materially degrade? Return JSON with classification TOKEN_ESSENTIAL, TOKEN_STRONGLY_JUSTIFIED, TOKEN_OPTIONAL, or TOKEN_UNNECESSARY; score 0-100; rationale; fiat_or_database_substitution. Proposal: " + json.dumps(proposal.pitch, default=str) + " Crypto: " + json.dumps(crypto, default=str)) if sol_review: score = float(sol_review.get("score", score)) @@ -385,7 +542,57 @@ class CryptoVentureService: classification = TokenNecessityClassification.TOKEN_UNNECESSARY if sol_classification in TokenNecessityClassification.values: classification = sol_classification - return TokenUtilityAssessment.objects.update_or_create(proposal=proposal, defaults={"classification": classification, "token_necessity_score": score, "utility_categories": utilities, "fiat_or_database_substitution": str((sol_review or {}).get("fiat_or_database_substitution", "If normal fiat/stablecoin/database credits preserve the core coordination, route to SaaS.")), "rationale": str((sol_review or {}).get("rationale", "Token score is based on explicit non-speculative utility, staking/slashing, marketplace coordination, and usage-linked fee demand.")), "metadata": {"weak_utility_terms": INSUFFICIENT_TOKEN_UTILITIES, "sol_counterfactual_review": sol_review or {}}})[0] + 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 {}, "required_roles": roles.required_roles, "payment_only_penalty": payment_only}})[0] + + def token_role_decomposition(self, proposal: CompanyProposal) -> TokenRoleDecomposition: + text = self._proposal_text(proposal) + roles = {} + role_terms = { + "PAYMENT": ["fee", "payment", "settlement"], + "SECURITY_BOND": ["bond", "security budget"], + "SLASHABLE_COLLATERAL": ["slash", "slashing", "collateral"], + "GOVERNANCE": ["governance", "vote"], + "RESOURCE_ACCESS": ["access", "scarce", "quota"], + "REPUTATION": ["reputation", "attestation"], + "REWARD": ["reward", "provider reward"], + "TREASURY": ["treasury"], + "OTHER": [], + } + for role, terms in role_terms.items(): + if role == "PAYMENT" and any(term in text for term in terms): + value = TokenRoleRequirement.USEFUL + 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 + elif any(term in text for term in terms): + value = TokenRoleRequirement.REQUIRED if role in {"SECURITY_BOND", "SLASHABLE_COLLATERAL", "REPUTATION"} else TokenRoleRequirement.USEFUL + else: + value = TokenRoleRequirement.UNNECESSARY + roles[role] = value + required = [role for role, value in roles.items() if value == TokenRoleRequirement.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_centralized_database_credits": "routes to SaaS if verification/settlement/reputation do not degrade", "usdc_identical_payment_utility": roles.get("PAYMENT") != TokenRoleRequirement.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." + return TokenRoleDecomposition.objects.update_or_create(proposal=proposal, defaults={"roles": roles, "required_roles": required, "stablecoin_counterfactual": stablecoin, "native_token_removed_outcome": outcome})[0] + + def assess_pretoken_monetization(self, proposal: CompanyProposal) -> PreTokenMonetizationAssessment: + 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 + if "membership" in text or "founding" in text: + mode = PreTokenMonetizationMode.FOUNDING_MEMBERSHIP_NFT + immediate = ["testnet/API access", "usage credits", "priority queues", "protocol analytics"] if mode != PreTokenMonetizationMode.NONE else [] + score = 0 if mode == PreTokenMonetizationMode.NONE else 55 + (15 if "api" in text else 0) + (10 if "automated" in text or "machine" in text else 0) + score = min(100, score) + membership_policy = {"no_investment_return": True, "no_equity": True, "no_profit_or_revenue_share": True, "no_guaranteed_native_token_allocation": True, "no_appreciation_promise": True, "useful_without_future_token": True, "public_sale_requires_human_legal_gate": True, "v02_sale_executed": False} + ladder = {"1000": "sell small paid beta/API-credit packages", "5000": "10-25 paid beta users or usage-credit customers", "10000": "recurring API/subscription/service-credit revenue without native token"} + return PreTokenMonetizationAssessment.objects.update_or_create(proposal=proposal, defaults={"mode": mode, "pre_token_monetization_potential": score, "pre_token_business": "Sell useful product/network access before native token issuance using fiat/stablecoin credits, paid beta, subscription, or legally reviewed membership. No native token required.", "immediate_utility": immediate, "revenue_ladder": ladder, "founding_membership_policy": membership_policy, "autonomous_fulfillment_notes": "Prefer self-service API/testnet onboarding, automated usage metering, and docs-first support."})[0] + + def token_economic_model(self, proposal: CompanyProposal) -> TokenEconomicModel: + text = self._proposal_text(proposal) + token = getattr(proposal, "token_utility_assessment", None) or self.assess_token_necessity(proposal) + avg_fee = round(0.03 + token.token_necessity_score / 1200 + (0.04 if "api" in text else 0), 3) + providers = max(3, 4 + len(token.utility_categories) * 2) + users = max(25, int(float(proposal.confidence) * 220)) + stake = 200 + int(token.token_necessity_score * 10) + return TokenEconomicModel.objects.update_or_create(proposal=proposal, defaults={"unit_of_service": "verified job/API call/attestation", "expected_usage_frequency": "weekly to daily machine/API usage", "transaction_fee_model": "Per-use stablecoin or credit fee pre-token; protocol fees may later settle through approved token design.", "average_fee": avg_fee, "payment_asset": "stablecoin_or_fiat_pre_token", "stake_requirement": stake, "collateral_requirement": stake, "provider_count": providers, "demand_side_users": users, "slash_event_assumptions": {"baseline_rate": 0.01, "high_slash_rate": 0.08, "objective_conditions_required": True}, "slash_amount": round(stake * 0.15, 2), "emission_schedule": {"v02": "none", "post_launch_model_only": "declining bootstrap subsidies"}, "bootstrap_subsidies": 0.0, "treasury_share": 0.2, "provider_reward_share": 0.8, "token_sinks": ["provider bonds", "slashing penalties", "protocol fees", "security budget"], "unlock_assumptions": {"v02": "none"}, "circulation_assumptions": {"v02_native_token_supply": 0}})[0] def _create_proposal(self, mandate: CompanyMandate, payload: dict[str, Any], source: str, territory: str) -> CompanyProposal: title = str(payload.get("name") or payload.get("title") or f"Protocol {territory.title()}")[:255] @@ -435,6 +642,7 @@ class CryptoVentureService: onchain = self.onchain_assessment(proposal) value = self.value_capture_assessment(proposal) policy = self.regulatory_policy(proposal) + roles = self.token_role_decomposition(proposal) flags = [] if token.classification in {TokenNecessityClassification.TOKEN_OPTIONAL, TokenNecessityClassification.TOKEN_UNNECESSARY}: flags.append(TokenRedTeamFlag.TOKEN_NOT_REQUIRED) @@ -451,7 +659,26 @@ class CryptoVentureService: flags.append(TokenRedTeamFlag.SPECULATION_DEPENDENT) if "governance" in text and "slash" not in text and "fee" not in text: flags.append(TokenRedTeamFlag.GOVERNANCE_THEATER) - sol_review = self._sol_json("Independent Token Red Team. Return JSON with flags list using only TOKEN_NOT_REQUIRED, SPECULATION_DEPENDENT, UNSUSTAINABLE_EMISSIONS, VALUE_CAPTURE_BROKEN, MERCENARY_INCENTIVES, GOVERNANCE_THEATER, SECURITY_MODEL_WEAK, TOKEN_VELOCITY_TOO_HIGH, BOOTSTRAP_PROBLEM, CENTRALIZATION_CONTRADICTION, REGULATORY_RISK_HIGH, ONCHAIN_NOT_REQUIRED, NO_REAL_USER_DEMAND; severity LOW/MEDIUM/HIGH; critique. Proposal: " + json.dumps(self.crypto_score_context(proposal), default=str)) + if len([role for role, value in roles.roles.items() if value in {TokenRoleRequirement.REQUIRED, TokenRoleRequirement.USEFUL}]) >= 6: + flags.append(TokenRedTeamFlag.TOKEN_DOES_TOO_MUCH) + if roles.roles.get("PAYMENT") in {TokenRoleRequirement.USEFUL, TokenRoleRequirement.OPTIONAL}: + flags.append(TokenRedTeamFlag.NATIVE_PAYMENT_NOT_REQUIRED) + if roles.stablecoin_counterfactual.get("stable_collateral_identical_security"): + flags.append(TokenRedTeamFlag.STABLECOIN_SUBSTITUTE_WORKS) + flags.append(TokenRedTeamFlag.COLLATERAL_CAN_BE_EXTERNAL) + if any(term in text for term in ["buyback", "revenue share", "profit share"]): + flags.append(TokenRedTeamFlag.REVENUE_CLAIM_LANGUAGE) + if "buyback" in text: + flags.append(TokenRedTeamFlag.BUYBACK_DEPENDENCY) + if "yield" in text or "apy" in text: + flags.append(TokenRedTeamFlag.YIELD_DEPENDENCY) + if "manual dispute" in text or "human court" in text: + flags.append(TokenRedTeamFlag.HUMAN_DISPUTE_DEPENDENCY) + if "central operator" in text: + flags.append(TokenRedTeamFlag.CENTRAL_OPERATOR_CONTRADICTION) + if "subjective" in text and "slashing" in text: + flags.append(TokenRedTeamFlag.SLASHING_NOT_OBJECTIVELY_MEASURABLE) + 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)) if sol_review: for flag in self._as_list(sol_review.get("flags")): if str(flag) in TokenRedTeamFlag.values: @@ -489,26 +716,27 @@ class CryptoVentureService: def simulate_tokenomics(self, proposal: CompanyProposal) -> TokenomicsSimulation: scenarios = {} - text = self._proposal_text(proposal) - token = getattr(proposal, "token_utility_assessment", None) or self.assess_token_necessity(proposal) - value = getattr(proposal, "value_capture", None) or self.value_capture_assessment(proposal) - base_users = max(20, int(float(proposal.confidence) * 180)) - fee_per_tx = 0.04 + (token.token_necessity_score / 1000) + (0.03 if "enterprise" in text or "infrastructure" in text else 0) - tx_per_user = 6 + len(token.utility_categories) * 2 + (4 if "agent" in text or "machine" in text else 0) - staking_per_provider = 150 + int(value.value_accrual_quality_score * 8) - emission_rate = max(0.005, (100 - value.tokenomics_sustainability_score) / 1400) - for name, multiplier in {"LOW_USAGE": 0.3, "EXPECTED": 1.0, "HIGH_USAGE": 3.0, "SUBSIDY_REMOVED": 0.8, "TOKEN_PRICE_DOWN_80_PERCENT": 0.8, "PROVIDER_CHURN": 0.6, "USER_GROWTH_10X": 10.0}.items(): + model = getattr(proposal, "token_economic_model", None) or self.token_economic_model(proposal) + roles = getattr(proposal, "token_role_decomposition", None) or self.token_role_decomposition(proposal) + base_users = max(1, model.demand_side_users) + tx_per_user = 10 + scenario_multipliers = {"LOW_USAGE": 0.3, "EXPECTED_USAGE": 1.0, "HIGH_USAGE": 3.0, "NO_SUBSIDY": 0.9, "TOKEN_PRICE_DOWN_80_PERCENT": 0.8, "TOKEN_PRICE_UP_10X": 1.2, "PROVIDER_CHURN_50_PERCENT": 0.7, "USER_GROWTH_10X": 10.0, "HIGH_SLASH_RATE": 1.0, "ZERO_NEW_TOKEN_EMISSIONS": 1.0, "STABLECOIN_PAYMENT_SUBSTITUTE": 1.0, "NATIVE_TOKEN_REMOVED": 1.0} + for name, multiplier in scenario_multipliers.items(): users = int(base_users * multiplier) tx = int(users * tx_per_user) - fees = round(tx * fee_per_tx, 2) - provider_supply = max(3, int(users / (15 if "market" in text else 25))) - staking = provider_supply * staking_per_provider - emissions = 0 if name == "SUBSIDY_REMOVED" else round(max(3, tx * emission_rate), 2) - if name == "PROVIDER_CHURN": + fees = round(tx * model.average_fee, 2) + provider_supply = max(1, int(model.provider_count * multiplier)) + staking = provider_supply * model.stake_requirement + slash_rate = float(model.slash_event_assumptions.get("high_slash_rate" if name == "HIGH_SLASH_RATE" else "baseline_rate", 0.01)) + emissions = 0 if name in {"NO_SUBSIDY", "ZERO_NEW_TOKEN_EMISSIONS"} else round(max(0, tx * 0.01), 2) + if name == "PROVIDER_CHURN_50_PERCENT": provider_supply = max(1, int(provider_supply * 0.45)) staking = max(100, int(staking * 0.45)) - scenarios[name] = {"users": users, "transactions": tx, "fees": fees, "token_demand": fees + staking * 0.01, "provider_supply": provider_supply, "staking": staking, "emissions": emissions, "treasury": round(1000 + fees * 0.2 - emissions * 0.1, 2), "circulating_supply": 1_000_000 + emissions, "token_velocity": round(tx / max(1, fees + staking * 0.01), 2), "reward_coverage": round(fees / max(1, emissions), 2), "network_security_budget": staking} - summary = {"token_price_appreciation_primary_success_variable": False, "subsidy_removed_survives": scenarios["SUBSIDY_REMOVED"]["reward_coverage"] >= 1.0, "stress_notes": ["Price-down and provider-churn scenarios must preserve service quality before mainnet."]} + native_removed = name == "NATIVE_TOKEN_REMOVED" + stablecoin_substitute = name == "STABLECOIN_PAYMENT_SUBSTITUTE" + protocol_degrades = native_removed and bool(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, "protocol_degrades_without_native_token": protocol_degrades} + 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_SUBSTITUTE"]["stablecoin_payment_works"], "stress_notes": ["Price appreciation is not a success variable.", "If native-token-removed changes little, reduce token necessity."]} return TokenomicsSimulation.objects.update_or_create(proposal=proposal, defaults={"scenarios": scenarios, "summary": summary})[0] def crypto_scenario_lab(self, cohort: VentureCohort) -> list[CryptoScenarioLab]: @@ -525,6 +753,9 @@ class CryptoVentureService: policy = self.regulatory_policy(proposal) red = getattr(proposal, "crypto_red_team", None) or self.red_team_proposal(proposal) sim = getattr(proposal, "tokenomics_simulation", None) or self.simulate_tokenomics(proposal) + pretoken = getattr(proposal, "pretoken_monetization", None) or self.assess_pretoken_monetization(proposal) + roles = getattr(proposal, "token_role_decomposition", None) or self.token_role_decomposition(proposal) + security = getattr(proposal, "protocol_security_assessment", None) autonomy = self.assess_autonomy(proposal) text = self._proposal_text(proposal) component_scores = { @@ -538,10 +769,12 @@ class CryptoVentureService: "AUTONOMOUS_OPERABILITY": autonomy.autonomous_operability_score, "SECURITY_MODEL_QUALITY": min(100, 45 + sum(8 for term in ["contract", "oracle", "key", "pause", "treasury", "slashing"] if term in text)), "REGULATORY_MANAGEABILITY": policy.regulatory_manageability_score, + "PRE_TOKEN_MONETIZATION_POTENTIAL": pretoken.pre_token_monetization_potential, } score = round(sum(component_scores.values()) / len(component_scores) - len(red.flags) * 4, 1) decision = self._crypto_decision(component_scores, token.classification, red.flags) - return {"proposal_id": str(proposal.id), "company": proposal.title, "one_line_thesis": proposal.pitch.get("One-line thesis", ""), "product_thesis": proposal.protocol_thesis.product_thesis, "protocol_thesis": proposal.protocol_thesis.protocol_thesis, "token_thesis": proposal.protocol_thesis.token_thesis, "why_onchain": proposal.onchain_assessment.rationale, "why_token": token.rationale, "token_utility": token.utility_categories, "token_necessity_classification": token.classification, "token_demand_loop": proposal.token_demand_loop.loop, "value_capture": proposal.value_capture.value_accrual, "network_effect": proposal.protocol_thesis.network_effect, "bootstrap_plan": proposal.protocol_thesis.bootstrap_plan, "autonomous_operability": autonomy.autonomous_operability_score, "regulatory_manageability": policy.regulatory_manageability_score, "security_risk": component_scores["SECURITY_MODEL_QUALITY"], "validation_experiment": proposal.validation_plan, "component_scores": component_scores, "value_accrual_quality": value.value_accrual_quality_score, "tokenomics_sustainability": value.tokenomics_sustainability_score, "simulation_summary": sim.summary, "token_red_team_flags": red.flags, "crypto_ic_decision": decision, "crypto_ic_score": score, "research": proposal.metadata.get("research", {}), "legal_review_required": policy.legal_review_required} + autonomy_class = CryptoAutonomyClass.AUTONOMOUS_CRYPTO if autonomy.autonomous_operability_score >= 70 else CryptoAutonomyClass.ASSISTED_CRYPTO + 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, "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"], "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: assessment = self.venture.assess_autonomous_operability(proposal) @@ -590,7 +823,7 @@ class CryptoVentureService: def _qualifies_finalist(self, row: dict[str, Any]) -> bool: scores = row["component_scores"] - return row["token_necessity_classification"] in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED} and scores["TOKEN_NECESSITY"] >= 75 and scores["REAL_USAGE_DEMAND"] >= 65 and scores["ONCHAIN_NECESSITY"] >= 70 and scores["VALUE_ACCRUAL_QUALITY"] >= 65 and scores["TOKENOMICS_SUSTAINABILITY"] >= 65 and scores["AUTONOMOUS_OPERABILITY"] >= 70 + 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 def _crypto_concentration(self, rows: list[dict[str, Any]]) -> dict[str, Any]: categories = Counter(row["company"].split()[0] for row in rows) @@ -661,7 +894,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) def _readable_report(self, content: dict[str, Any]) -> str: - lines = ["# CRYPTO VENTURE COHORT V0.1 REPORT", "", f"Cohort ID: {content['cohort_id']}", f"Accepted protocols: {content['accepted_protocols']}", "", "## Ranking"] + lines = ["# CRYPTO VENTURE COHORT V0.2", "", f"Cohort ID: {content['cohort_id']}", f"Crypto survivors: {content['crypto_survivors']}", "", "## 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("") @@ -671,5 +904,5 @@ class CryptoVentureService: else: lines.append("Fewer than 3 qualified; weak token ideas were not promoted.") lines.append("") - lines.append("Stop condition: no token sale, fundraising, mainnet issuance, user contact, or real spend.") + lines.append("Stop condition: no token/NFT sale, fundraising, mainnet issuance, user contact, or real spend.") return "\n".join(lines) diff --git a/control_plane/ventures/management/commands/export_crypto_venture_cohort.py b/control_plane/ventures/management/commands/export_crypto_venture_cohort.py index 138cb86..0c21a65 100644 --- a/control_plane/ventures/management/commands/export_crypto_venture_cohort.py +++ b/control_plane/ventures/management/commands/export_crypto_venture_cohort.py @@ -59,11 +59,13 @@ class Command(BaseCommand): def markdown(self, payload: dict[str, Any]) -> str: lines = [ - "# CRYPTO VENTURE COHORT V0.1 REPORT", + f"# {payload.get('title', 'CRYPTO VENTURE COHORT REPORT')}", "", f"Cohort ID: `{payload['cohort_id']}`", f"GraphRun: `{payload['graph_run']}`", - f"Accepted protocols: `{payload['accepted_protocols']}`", + f"Raw protocols generated: `{payload.get('raw_protocols_generated', payload.get('accepted_protocols', 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"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"Generation attempts: `{payload['generation_attempts']}`", "", "## Runtime", @@ -88,6 +90,16 @@ class Command(BaseCommand): "", f"Token thesis: {row['token_thesis']}", "", + f"Pre-token business: {row.get('pre_token_business', '')}", + "", + f"Pre-token monetization: `{row.get('pre_token_monetization', '')}`. Potential: `{row.get('pre_token_monetization_potential', '')}`. Revenue ladder: {json.dumps(row.get('pre_token_revenue_ladder', {}), default=str)}", + "", + f"Token role decomposition: {json.dumps(row.get('token_role_decomposition', {}), default=str)}", + "", + f"Stablecoin counterfactual: {json.dumps(row.get('stablecoin_counterfactual', {}), default=str)}", + "", + f"Native token removed outcome: {row.get('native_token_removed_outcome', '')}", + "", f"Token demand loop: {self.inline(row['token_demand_loop'])}", "", f"Value capture: {row['value_capture']}", @@ -96,7 +108,7 @@ class Command(BaseCommand): "", f"Bootstrap plan: {row['bootstrap_plan']}", "", - f"Autonomous operability: `{row['autonomous_operability']}`. Regulatory manageability: `{row['regulatory_manageability']}`. Security risk score: `{row['security_risk']}`.", + f"Autonomous operability: `{row['autonomous_operability']}`. Autonomy class: `{row.get('crypto_autonomy_class', '')}`. Regulatory manageability: `{row['regulatory_manageability']}`. Security: `{row.get('security_assessment', '')}`. Security risk score: `{row['security_risk']}`.", "", f"Scores: {json.dumps(row['component_scores'], default=str)}", "", @@ -112,7 +124,7 @@ class Command(BaseCommand): lines.extend(f"- {row['company']}: `{row['crypto_ic_decision']}`" for row in top_3) else: lines.append("Fewer than 3 qualified; weak token ideas were not promoted.") - lines.extend(["", "## Token Red Team Failures", "", json.dumps(payload.get("token_red_team_failures", {}), indent=2, default=str), "", "## Token Utility Distribution", "", json.dumps(payload.get("token_utility_distribution", {}), indent=2, default=str), "", "## Capability Gaps", "", json.dumps(payload.get("capability_gaps", []), indent=2, default=str), "", "Stop condition: no token sale, no fundraising, no mainnet issuance, no investor/user contact, no liquidity pool, no market making, no real spend."]) + lines.extend(["", "## Token Red Team Failures", "", json.dumps(payload.get("token_red_team_failures", {}), indent=2, default=str), "", "## Research Health", "", json.dumps(payload.get("research_health", {}), indent=2, default=str), "", "## Token Utility Distribution", "", json.dumps(payload.get("token_utility_distribution", {}), indent=2, default=str), "", "## Capability Gaps", "", json.dumps(payload.get("capability_gaps", []), indent=2, default=str), "", "Stop condition: no token sale, no NFT sale, no Founding Membership sale, no fundraising, no mainnet issuance, no investor/user contact, no liquidity pool, no market making, no real spend."]) return "\n".join(lines).rstrip() def inline(self, value: Any) -> str: diff --git a/control_plane/ventures/management/commands/run_crypto_venture_cohort.py b/control_plane/ventures/management/commands/run_crypto_venture_cohort.py index c5680e8..92c0df0 100644 --- a/control_plane/ventures/management/commands/run_crypto_venture_cohort.py +++ b/control_plane/ventures/management/commands/run_crypto_venture_cohort.py @@ -19,7 +19,7 @@ class Command(BaseCommand): help = "Run a Crypto / Protocol Venture Cohort V0.1." def add_arguments(self, parser): - parser.add_argument("--size", type=int, default=10) + parser.add_argument("--size", type=int, default=30) parser.add_argument("--concurrency", type=int, default=2) parser.add_argument("--qwen-only", action="store_true", help="Use Qwen for all crypto cohort model roles.") parser.add_argument("--sol-final-ic", action="store_true", help="Use Sol for final IC role when configured.") @@ -81,6 +81,13 @@ class Command(BaseCommand): "members": len(members), "concurrency": cohort.concurrency, "metrics": cohort.metrics, + "raw_protocols_generated": cohort.metrics.get("raw_generated", 0), + "token_unnecessary": cohort.metrics.get("token_unnecessary_rejections", 0), + "token_optional_routed_saas": cohort.metrics.get("token_optional_route_to_saas", 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": cohort.metrics.get("finalists", 0), "generation_sources": sorted({str(member.proposal.metadata.get("generation_source", "unknown")) for member in members}), "top_3": review.top_3 if review else [], "rejected_token_not_needed": [row for row in (review.rankings if review else []) if row.get("crypto_ic_decision") == "REJECT_TOKEN_NOT_NEEDED"], diff --git a/control_plane/ventures/migrations/0006_pretokenmonetizationassessment_and_more.py b/control_plane/ventures/migrations/0006_pretokenmonetizationassessment_and_more.py new file mode 100644 index 0000000..e217360 --- /dev/null +++ b/control_plane/ventures/migrations/0006_pretokenmonetizationassessment_and_more.py @@ -0,0 +1,130 @@ +# Generated by Django 5.2.16 on 2026-08-16 12:19 + +import django.db.models.deletion +import uuid +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('ventures', '0005_alter_autonomousoperabilityassessment_venture_track_and_more'), + ] + + operations = [ + migrations.CreateModel( + name='PreTokenMonetizationAssessment', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('mode', models.CharField(choices=[('NONE', 'None'), ('PAID_BETA_ACCESS', 'Paid Beta Access'), ('SERVICE_CREDITS', 'Service Credits'), ('STABLECOIN_USAGE_FEES', 'Stablecoin Usage Fees'), ('SUBSCRIPTION', 'Subscription'), ('FOUNDING_MEMBERSHIP', 'Founding Membership'), ('FOUNDING_MEMBERSHIP_NFT', 'Founding Membership Nft'), ('OTHER', 'Other')], default='NONE', max_length=40)), + ('pre_token_monetization_potential', models.FloatField(default=0.0)), + ('pre_token_business', models.TextField(blank=True)), + ('immediate_utility', models.JSONField(blank=True, default=list)), + ('revenue_ladder', models.JSONField(blank=True, default=dict)), + ('founding_membership_policy', models.JSONField(blank=True, default=dict)), + ('autonomous_fulfillment_notes', models.TextField(blank=True)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='pretoken_monetization', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='ProtocolSecurityAssessment', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('decision', models.CharField(choices=[('PASS_FOR_TESTNET_DESIGN', 'Pass For Testnet Design'), ('REVISE_SECURITY_MODEL', 'Revise Security Model'), ('BLOCK_TESTNET', 'Block Testnet')], max_length=40)), + ('threat_model', models.JSONField(blank=True, default=dict)), + ('missing_controls', models.JSONField(blank=True, default=list)), + ('privileged_roles', models.JSONField(blank=True, default=list)), + ('economic_attack_surfaces', models.JSONField(blank=True, default=list)), + ('key_management_assumptions', models.JSONField(blank=True, default=list)), + ('rationale', models.TextField(blank=True)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='protocol_security_assessment', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='TokenEconomicModel', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('unit_of_service', models.CharField(blank=True, max_length=160)), + ('expected_usage_frequency', models.CharField(blank=True, max_length=160)), + ('transaction_fee_model', models.TextField(blank=True)), + ('average_fee', models.FloatField(default=0.0)), + ('payment_asset', models.CharField(blank=True, max_length=80)), + ('stake_requirement', models.FloatField(default=0.0)), + ('collateral_requirement', models.FloatField(default=0.0)), + ('provider_count', models.PositiveIntegerField(default=0)), + ('demand_side_users', models.PositiveIntegerField(default=0)), + ('slash_event_assumptions', models.JSONField(blank=True, default=dict)), + ('slash_amount', models.FloatField(default=0.0)), + ('emission_schedule', models.JSONField(blank=True, default=dict)), + ('bootstrap_subsidies', models.FloatField(default=0.0)), + ('treasury_share', models.FloatField(default=0.0)), + ('provider_reward_share', models.FloatField(default=0.0)), + ('token_sinks', models.JSONField(blank=True, default=list)), + ('unlock_assumptions', models.JSONField(blank=True, default=dict)), + ('circulation_assumptions', models.JSONField(blank=True, default=dict)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='token_economic_model', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='TokenLaunchReadinessAssessment', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('real_users', models.BooleanField(default=False)), + ('repeat_usage', models.BooleanField(default=False)), + ('real_protocol_fees', models.BooleanField(default=False)), + ('token_necessity_validated', models.BooleanField(default=False)), + ('token_demand_loop_validated', models.BooleanField(default=False)), + ('security_audit_status', models.CharField(blank=True, max_length=120)), + ('testnet_stability', models.CharField(blank=True, max_length=120)), + ('tokenomics_stress_tests', models.JSONField(blank=True, default=dict)), + ('legal_review_status', models.CharField(blank=True, max_length=120)), + ('jurisdiction_policy_status', models.CharField(blank=True, max_length=120)), + ('admin_key_controls', models.CharField(blank=True, max_length=120)), + ('treasury_controls', models.CharField(blank=True, max_length=120)), + ('decentralization_readiness', models.CharField(blank=True, max_length=120)), + ('launch_readiness_status', models.CharField(choices=[('NOT_READY', 'Not Ready'), ('PRODUCT_TRACTION_REQUIRED', 'Product Traction Required'), ('TOKEN_MODEL_REVISION_REQUIRED', 'Token Model Revision Required'), ('SECURITY_REVIEW_REQUIRED', 'Security Review Required'), ('LEGAL_REVIEW_REQUIRED', 'Legal Review Required'), ('READY_FOR_HUMAN_LAUNCH_REVIEW', 'Ready For Human Launch Review')], default='NOT_READY', max_length=60)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='token_launch_readiness', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='TokenRoleDecomposition', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('roles', models.JSONField(blank=True, default=dict)), + ('required_roles', models.JSONField(blank=True, default=list)), + ('stablecoin_counterfactual', models.JSONField(blank=True, default=dict)), + ('native_token_removed_outcome', models.TextField(blank=True)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('proposal', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='token_role_decomposition', to='ventures.companyproposal')), + ], + options={ + 'abstract': False, + }, + ), + ] diff --git a/control_plane/ventures/models.py b/control_plane/ventures/models.py index de7886a..c59e58b 100644 --- a/control_plane/ventures/models.py +++ b/control_plane/ventures/models.py @@ -131,6 +131,59 @@ class TokenRedTeamFlag(models.TextChoices): REGULATORY_RISK_HIGH = "REGULATORY_RISK_HIGH" ONCHAIN_NOT_REQUIRED = "ONCHAIN_NOT_REQUIRED" NO_REAL_USER_DEMAND = "NO_REAL_USER_DEMAND" + TOKEN_DOES_TOO_MUCH = "TOKEN_DOES_TOO_MUCH" + NATIVE_PAYMENT_NOT_REQUIRED = "NATIVE_PAYMENT_NOT_REQUIRED" + STABLECOIN_SUBSTITUTE_WORKS = "STABLECOIN_SUBSTITUTE_WORKS" + COLLATERAL_CAN_BE_EXTERNAL = "COLLATERAL_CAN_BE_EXTERNAL" + REVENUE_CLAIM_LANGUAGE = "REVENUE_CLAIM_LANGUAGE" + BUYBACK_DEPENDENCY = "BUYBACK_DEPENDENCY" + YIELD_DEPENDENCY = "YIELD_DEPENDENCY" + SPECULATIVE_BOOTSTRAP = "SPECULATIVE_BOOTSTRAP" + EMISSIONS_DEPENDENCY = "EMISSIONS_DEPENDENCY" + SECURITY_TOKEN_CONCENTRATION = "SECURITY_TOKEN_CONCENTRATION" + SLASHING_NOT_OBJECTIVELY_MEASURABLE = "SLASHING_NOT_OBJECTIVELY_MEASURABLE" + ORACLE_CIRCULARITY = "ORACLE_CIRCULARITY" + HUMAN_DISPUTE_DEPENDENCY = "HUMAN_DISPUTE_DEPENDENCY" + CENTRAL_OPERATOR_CONTRADICTION = "CENTRAL_OPERATOR_CONTRADICTION" + NETWORK_NOT_READY_FOR_DECENTRALIZATION = "NETWORK_NOT_READY_FOR_DECENTRALIZATION" + + +class PreTokenMonetizationMode(models.TextChoices): + NONE = "NONE" + PAID_BETA_ACCESS = "PAID_BETA_ACCESS" + SERVICE_CREDITS = "SERVICE_CREDITS" + STABLECOIN_USAGE_FEES = "STABLECOIN_USAGE_FEES" + SUBSCRIPTION = "SUBSCRIPTION" + FOUNDING_MEMBERSHIP = "FOUNDING_MEMBERSHIP" + FOUNDING_MEMBERSHIP_NFT = "FOUNDING_MEMBERSHIP_NFT" + OTHER = "OTHER" + + +class TokenRoleRequirement(models.TextChoices): + REQUIRED = "REQUIRED" + USEFUL = "USEFUL" + OPTIONAL = "OPTIONAL" + UNNECESSARY = "UNNECESSARY" + + +class TokenLaunchReadinessStatus(models.TextChoices): + NOT_READY = "NOT_READY" + PRODUCT_TRACTION_REQUIRED = "PRODUCT_TRACTION_REQUIRED" + TOKEN_MODEL_REVISION_REQUIRED = "TOKEN_MODEL_REVISION_REQUIRED" + SECURITY_REVIEW_REQUIRED = "SECURITY_REVIEW_REQUIRED" + LEGAL_REVIEW_REQUIRED = "LEGAL_REVIEW_REQUIRED" + READY_FOR_HUMAN_LAUNCH_REVIEW = "READY_FOR_HUMAN_LAUNCH_REVIEW" + + +class ProtocolSecurityDecision(models.TextChoices): + PASS_FOR_TESTNET_DESIGN = "PASS_FOR_TESTNET_DESIGN" + REVISE_SECURITY_MODEL = "REVISE_SECURITY_MODEL" + BLOCK_TESTNET = "BLOCK_TESTNET" + + +class CryptoAutonomyClass(models.TextChoices): + AUTONOMOUS_CRYPTO = "AUTONOMOUS_CRYPTO" + ASSISTED_CRYPTO = "ASSISTED_CRYPTO" class FounderDependencyLevel(models.TextChoices): @@ -389,6 +442,81 @@ class CryptoScenarioLab(TimestampedModel): metadata = models.JSONField(default=dict, blank=True) +class PreTokenMonetizationAssessment(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="pretoken_monetization") + mode = models.CharField(max_length=40, choices=PreTokenMonetizationMode.choices, default=PreTokenMonetizationMode.NONE) + pre_token_monetization_potential = models.FloatField(default=0.0) + pre_token_business = models.TextField(blank=True) + immediate_utility = models.JSONField(default=list, blank=True) + revenue_ladder = models.JSONField(default=dict, blank=True) + founding_membership_policy = models.JSONField(default=dict, blank=True) + autonomous_fulfillment_notes = models.TextField(blank=True) + metadata = models.JSONField(default=dict, blank=True) + + +class TokenRoleDecomposition(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="token_role_decomposition") + roles = models.JSONField(default=dict, blank=True) + required_roles = models.JSONField(default=list, blank=True) + stablecoin_counterfactual = models.JSONField(default=dict, blank=True) + native_token_removed_outcome = models.TextField(blank=True) + metadata = models.JSONField(default=dict, blank=True) + + +class TokenEconomicModel(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="token_economic_model") + unit_of_service = models.CharField(max_length=160, blank=True) + expected_usage_frequency = models.CharField(max_length=160, blank=True) + transaction_fee_model = models.TextField(blank=True) + average_fee = models.FloatField(default=0.0) + payment_asset = models.CharField(max_length=80, blank=True) + stake_requirement = models.FloatField(default=0.0) + collateral_requirement = models.FloatField(default=0.0) + provider_count = models.PositiveIntegerField(default=0) + demand_side_users = models.PositiveIntegerField(default=0) + slash_event_assumptions = models.JSONField(default=dict, blank=True) + slash_amount = models.FloatField(default=0.0) + emission_schedule = models.JSONField(default=dict, blank=True) + bootstrap_subsidies = models.FloatField(default=0.0) + treasury_share = models.FloatField(default=0.0) + provider_reward_share = models.FloatField(default=0.0) + token_sinks = models.JSONField(default=list, blank=True) + unlock_assumptions = models.JSONField(default=dict, blank=True) + circulation_assumptions = models.JSONField(default=dict, blank=True) + metadata = models.JSONField(default=dict, blank=True) + + +class TokenLaunchReadinessAssessment(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="token_launch_readiness") + real_users = models.BooleanField(default=False) + repeat_usage = models.BooleanField(default=False) + real_protocol_fees = models.BooleanField(default=False) + token_necessity_validated = models.BooleanField(default=False) + token_demand_loop_validated = models.BooleanField(default=False) + security_audit_status = models.CharField(max_length=120, blank=True) + testnet_stability = models.CharField(max_length=120, blank=True) + tokenomics_stress_tests = models.JSONField(default=dict, blank=True) + legal_review_status = models.CharField(max_length=120, blank=True) + jurisdiction_policy_status = models.CharField(max_length=120, blank=True) + admin_key_controls = models.CharField(max_length=120, blank=True) + treasury_controls = models.CharField(max_length=120, blank=True) + decentralization_readiness = models.CharField(max_length=120, blank=True) + launch_readiness_status = models.CharField(max_length=60, choices=TokenLaunchReadinessStatus.choices, default=TokenLaunchReadinessStatus.NOT_READY) + metadata = models.JSONField(default=dict, blank=True) + + +class ProtocolSecurityAssessment(TimestampedModel): + proposal = models.OneToOneField(CompanyProposal, on_delete=models.CASCADE, related_name="protocol_security_assessment") + decision = models.CharField(max_length=40, choices=ProtocolSecurityDecision.choices) + threat_model = models.JSONField(default=dict, blank=True) + missing_controls = models.JSONField(default=list, blank=True) + privileged_roles = models.JSONField(default=list, blank=True) + economic_attack_surfaces = models.JSONField(default=list, blank=True) + key_management_assumptions = models.JSONField(default=list, blank=True) + rationale = models.TextField(blank=True) + metadata = models.JSONField(default=dict, blank=True) + + class CompanyBoardReview(TimestampedModel): proposal = models.ForeignKey(CompanyProposal, on_delete=models.CASCADE, related_name="board_reviews") observations = models.JSONField(default=dict, blank=True) diff --git a/graph/crypto_venture_cohort.py b/graph/crypto_venture_cohort.py index 734f9c3..2b22861 100644 --- a/graph/crypto_venture_cohort.py +++ b/graph/crypto_venture_cohort.py @@ -30,7 +30,7 @@ def crypto_venture_cohort_graph_v1() -> ExecutionGraphSpec: "produce_crypto_cohort_report", "complete", ] - spec = ExecutionGraphSpec(name="crypto_venture_cohort", version=1, 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.1: token necessity, onchain necessity, token red team, simulation, regulatory gate, no issuance."}) + spec = ExecutionGraphSpec(name="crypto_venture_cohort", version=2, 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.2: 30-raw strict funnel, pre-token monetization, hardened research, token economic model, Guard-style security design review."}) spec.validate() return spec diff --git a/tests/test_crypto_venture_cohort.py b/tests/test_crypto_venture_cohort.py index 03548e6..cd7b621 100644 --- a/tests/test_crypto_venture_cohort.py +++ b/tests/test_crypto_venture_cohort.py @@ -50,9 +50,9 @@ def test_crypto_track_and_token_gate_remove_weak_then_regenerate() -> None: weak.refresh_from_db() assert weak.status == CompanyProposalStatus.REJECTED - assert weak.metadata["routed_to_saas"] is True - assert cohort.members.count() == 2 - assert all(member.proposal.token_utility_assessment.classification in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED} for member in cohort.members.select_related("proposal")) + assert weak.metadata["routed_to_saas"] is False + assert cohort.members.count() == 0 + assert cohort.metrics["regeneration_policy"].startswith("V0.2") def test_sol_counterfactual_red_team_and_final_ic_are_used() -> None: @@ -93,8 +93,28 @@ def test_tokenomics_simulation_is_proposal_specific() -> None: low_sim = svc.simulate_tokenomics(low) high_sim = svc.simulate_tokenomics(high) - assert low_sim.scenarios["EXPECTED"]["users"] != high_sim.scenarios["EXPECTED"]["users"] - assert low_sim.scenarios["EXPECTED"]["token_demand"] != high_sim.scenarios["EXPECTED"]["token_demand"] + assert low_sim.scenarios["EXPECTED_USAGE"]["users"] != high_sim.scenarios["EXPECTED_USAGE"]["users"] + assert low_sim.scenarios["EXPECTED_USAGE"]["token_demand"] != high_sim.scenarios["EXPECTED_USAGE"]["token_demand"] + assert "NATIVE_TOKEN_REMOVED" in low_sim.scenarios + + +def test_v02_pretoken_roles_security_and_launch_readiness() -> None: + svc = service() + cohort = svc.prepare_cohort(size=1, concurrency=1) + proposal = svc._create_proposal(cohort.mandate, svc._fallback_payload(0, "DECENTRALIZED_AI_COMPUTE"), "test", "DECENTRALIZED_AI_COMPUTE") + VentureCohortMember.objects.create(cohort=cohort, proposal=proposal) + + pretoken = svc.assess_pretoken_monetization(proposal) + roles = svc.token_role_decomposition(proposal) + model = svc.token_economic_model(proposal) + svc.protocol_security_gate(cohort) + readiness = svc.token_launch_readiness(proposal) + + assert pretoken.pre_token_monetization_potential > 0 + assert roles.stablecoin_counterfactual["MODEL_B_stablecoin_payment_native_bond"] + assert model.average_fee > 0 + assert proposal.protocol_security_assessment.decision in {"PASS_FOR_TESTNET_DESIGN", "REVISE_SECURITY_MODEL", "BLOCK_TESTNET"} + assert readiness.launch_readiness_status != "READY_FOR_HUMAN_LAUNCH_REVIEW" def test_crypto_cohort_graph_real_gates_and_report() -> None: