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