1153 lines
115 KiB
Python
1153 lines
115 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",
|
|
"VERIFIABLE_INFERENCE_NETWORKS",
|
|
"MACHINE_SERVICE_MARKETS",
|
|
"AGENT_ECONOMIC_IDENTITY",
|
|
"AUTONOMOUS_AGENT_COMMERCE",
|
|
"DECENTRALIZED_SECURITY_MARKETS",
|
|
"SOFTWARE_SECURITY_BONDS",
|
|
"PROOF_MARKETS",
|
|
"DECENTRALIZED_DATA_CONTRIBUTION",
|
|
"MODEL_CONTRIBUTION_NETWORKS",
|
|
"COMPUTE_CONTRIBUTION_NETWORKS",
|
|
"DECENTRALIZED_RESOURCE_MARKETS",
|
|
"BANDWIDTH_STORAGE_RESOURCE_MARKETS",
|
|
"DECENTRALIZED_INSURANCE_RISK_POOLS",
|
|
"PROTOCOLIZED_ESCROW",
|
|
"MACHINE_REPUTATION_WITH_ECONOMIC_STAKE",
|
|
"PROVIDER_COORDINATION_NETWORKS",
|
|
"AUTONOMOUS_INFRASTRUCTURE_MARKETS",
|
|
"OPEN_CRYPTO_CATEGORY",
|
|
]
|
|
|
|
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", "TOKEN_UTILITY_PRECEDENTS", "SECURITY_BOND_PRECEDENTS", "SLASHING_PRECEDENTS", "RESOURCE_MARKET_PRECEDENTS", "PROVIDER_ECONOMICS", "NETWORK_BOOTSTRAP", "VALUE_CAPTURE", "TOKENOMICS_FAILURES", "SECURITY_INCIDENTS", "ONCHAIN_ALTERNATIVES", "OFFCHAIN_ALTERNATIVES", "LEGAL_REGULATORY", "PRE_TOKEN_MONETIZATION"]
|
|
|
|
STRUCTURAL_TOKEN_ROLES = {"SECURITY_BOND", "SLASHABLE_COLLATERAL", "RESOURCE_ALLOCATION", "MACHINE_ECONOMIC_IDENTITY", "PROVIDER_ADMISSION", "DECENTRALIZED_SUPPLY_COORDINATION"}
|
|
WEAK_TOKEN_ROLES = {"DEMAND_SIDE_PAYMENT", "GOVERNANCE", "PROVIDER_REWARD", "TREASURY", "ACCESS"}
|
|
TOKEN_UTILITY_SCORE_KEYS = ["SECURITY_UTILITY", "COORDINATION_UTILITY", "COLLATERAL_UTILITY", "SLASHING_UTILITY", "RESOURCE_ALLOCATION_UTILITY", "INCENTIVE_UTILITY", "NETWORK_BOOTSTRAP_UTILITY", "VALUE_CAPTURE_UTILITY", "MACHINE_ECONOMIC_UTILITY", "GOVERNANCE_UTILITY", "PAYMENT_UTILITY", "SPECULATION_DEPENDENCE"]
|
|
|
|
PRIMARY_SOURCE_DOMAINS = ["ethereum.org", "eips.ethereum.org", "github.com", "arxiv.org", "sec.gov", "cftc.gov", "immunefi.com"]
|
|
SOURCE_REJECTION_REASONS = {"IRRELEVANT": "SOURCE_REJECT_IRRELEVANT", "WRONG_CATEGORY": "SOURCE_REJECT_WRONG_CATEGORY", "LOW_AUTHORITY": "SOURCE_REJECT_LOW_AUTHORITY", "SEARCH_NOISE": "SOURCE_REJECT_SEARCH_NOISE", "UNPARSABLE": "SOURCE_REJECT_UNPARSABLE", "DUPLICATE": "SOURCE_REJECT_DUPLICATE"}
|
|
|
|
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.3.1", "venture_track": VentureTrack.CRYPTO_PROTOCOL},
|
|
)
|
|
self._artifact(None, mandate, "CRYPTO_VENTURE_MANDATE", "Crypto Protocol Venture V0.3.1 Mandate", mandate.constraints, "Crypto protocol cohort mandate. Sol-native problem-first ideation and strong native token utility search. No token sale, fundraising, or mainnet issuance.", "crypto_venture")
|
|
return mandate
|
|
|
|
def prepare_cohort(self, *, size: int, graph_run=None, concurrency: int = 2) -> VentureCohort:
|
|
mandate = self.create_crypto_mandate()
|
|
cohort_id = f"CPV031-{time.strftime('%Y%m%d%H%M%S')}-{uuid.uuid4().hex[:8]}"
|
|
return VentureCohort.objects.create(cohort_id=cohort_id, mandate=mandate, cohort_size=size, graph_run=graph_run, concurrency=concurrency, status="PREPARING_CRYPTO", graph_versions={"cohort": "crypto_venture_cohort v3.1"}, research_policy={"source_linked_evidence_required": True, "research_insufficient_if_no_sources": True, "primary_source_first": True, "minimum_category_evidence": "one strong source or multiple independent medium sources", "crypto_categories": CRYPTO_RESEARCH_CATEGORIES}, scoring_policy={"dimensions": CRYPTO_SCORE_DIMENSIONS, "strict_token_filter": True, "sol_problem_first_generation": True, "strong_native_token_utility_required": True, "structural_roles": sorted(STRUCTURAL_TOKEN_ROLES)}, metadata={"milestone": "CRYPTO_PROTOCOL_VENTURE_COHORT_V0.3.1", "venture_track": VentureTrack.CRYPTO_PROTOCOL, "real_spend": 0, "real_customer_outreach": False, "token_sale": False, "fundraising": False, "mainnet_issuance": False, "v02_baseline": {"raw": 30, "crypto_survivors": 2, "autonomous_survivors": 0, "finalists": 0}, "v03_baseline": {"raw": 20, "onchain_rejected": 5, "onchain_passed": 15, "token_unnecessary": 12, "token_optional": 3, "crypto_survivors": 0, "finalists": 0}})
|
|
|
|
def generate_protocols(self, cohort: VentureCohort) -> list[CompanyProposal]:
|
|
accepted = []
|
|
attempts = 0
|
|
model_usage = Counter(cohort.metrics.get("model_usage", {}))
|
|
for index in range(cohort.cohort_size):
|
|
territory = CRYPTO_TERRITORIES[index % len(CRYPTO_TERRITORIES)]
|
|
payload, source, usage = self._protocol_payload(index, territory)
|
|
model_usage.update(usage)
|
|
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_theses": len(accepted), "raw_target": cohort.cohort_size, "raw_generated": len(accepted), "raw_sol_theses": len([p for p in accepted if p.metadata.get("generation_source") == "sol"]), "requested_protocol_count": cohort.cohort_size, "generation_attempts": attempts, "accepted_protocols": len(accepted), "duplicate_rejections": 0, "token_necessity_rejections": 0, "generation_sources": sorted({p.metadata.get("generation_source", "unknown") for p in accepted}), "model_usage": dict(model_usage), "sol_generation_requests": model_usage.get("sol_generation_requests", 0), "sol_onchain_judge_requests": model_usage.get("sol_onchain_judge_requests", 0), "fallback_count": model_usage.get("fallback_count", 0)}
|
|
cohort.status = "PROTOCOLS_GENERATED"
|
|
cohort.save(update_fields=["metrics", "status", "updated_at"])
|
|
return accepted
|
|
|
|
def onchain_necessity_gate(self, cohort: VentureCohort) -> list[OnchainNecessityAssessment]:
|
|
assessments = [self.onchain_assessment(member.proposal) for member in cohort.members.select_related("proposal")]
|
|
removed = 0
|
|
for member in list(cohort.members.select_related("proposal")):
|
|
proposal = member.proposal
|
|
assessment = getattr(proposal, "onchain_assessment", None) or self.onchain_assessment(proposal)
|
|
if assessment.onchain_necessity_score < 70:
|
|
VentureGenerationRejection.objects.create(cohort=cohort, slot_index=member.rank or 0, attempt=1, decision=NoveltyGateDecision.REGENERATE_HARD_EXCLUSION, reason="removed by onchain necessity gate", candidate={"title": proposal.title, "onchain_necessity_score": assessment.onchain_necessity_score}, similarity_score=0.0)
|
|
member.delete()
|
|
proposal.status = CompanyProposalStatus.REJECTED
|
|
proposal.metadata = {**proposal.metadata, "routed_to_saas": True, "routed_to_venture_studio": True, "onchain_rejected": True, "crypto_failure_reason": "onchain necessity failed"}
|
|
proposal.save(update_fields=["status", "metadata", "updated_at"])
|
|
removed += 1
|
|
onchain_passed = cohort.members.count()
|
|
cohort.metrics = {**cohort.metrics, "onchain_rejected": cohort.metrics.get("onchain_rejected", 0) + removed, "onchain_passed": onchain_passed, "crypto_survivors_after_onchain": onchain_passed, "token_design_skipped_due_to_onchain_rejection": cohort.metrics.get("token_design_skipped_due_to_onchain_rejection", 0) + removed}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
return assessments
|
|
|
|
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", "api reliability slashing pool", "verimesh", "verichain"]
|
|
for member in list(cohort.members.select_related("proposal").order_by("created_at")):
|
|
proposal = member.proposal
|
|
text = self._proposal_text(proposal)
|
|
utility_text = " ".join(json.dumps(item, sort_keys=True, default=str) if isinstance(item, dict) else str(item) for item in proposal.protocol_thesis.utility_categories)
|
|
fp = self._fingerprint(proposal.protocol_thesis.protocol_category + proposal.protocol_thesis.protocol_thesis + utility_text)
|
|
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.3.1 uses total raw Sol generation budget; offchain or 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]:
|
|
model_usage = Counter(cohort.metrics.get("model_usage", {}))
|
|
before = cohort.members.count()
|
|
assessments = []
|
|
for member in cohort.members.select_related("proposal"):
|
|
if not member.proposal.metadata.get("crypto", {}).get("token_design_completed"):
|
|
used = self.derive_token_design(member.proposal)
|
|
model_usage.update(used)
|
|
assessments.append(self.assess_token_necessity(member.proposal))
|
|
model_usage["sol_token_utility_judge_requests"] += 1 if self.final_ic_model_hint == "sol" else 0
|
|
removed = self._remove_weak_token_members(cohort)
|
|
cohort.metrics = {**cohort.metrics, **removed, "token_design_attempts": cohort.metrics.get("token_design_attempts", 0) + before, "token_design_count_matches_onchain_passed": (cohort.metrics.get("token_design_attempts", 0) + before) == cohort.metrics.get("onchain_passed", before), "model_usage": dict(model_usage), "sol_token_design_requests": model_usage.get("sol_token_design_requests", 0), "sol_token_utility_judge_requests": model_usage.get("sol_token_utility_judge_requests", 0)}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
return assessments
|
|
|
|
def derive_token_design(self, proposal: CompanyProposal) -> Counter:
|
|
usage = Counter()
|
|
crypto = proposal.metadata.get("crypto", {})
|
|
passes = crypto.get("generation_passes", {}) if isinstance(crypto, dict) else {}
|
|
pass_a = passes.get("pass_a_problem_network", {}) if isinstance(passes, dict) else {}
|
|
pass_b = passes.get("pass_b_onchain", {}) if isinstance(passes, dict) else {}
|
|
pass_c_prompt = "PASS C - MINIMAL DECENTRALIZED PROTOCOL ARCHITECTURE. Return JSON with protocol_architecture, participant_model, verification_model, settlement_model, resource_allocation_model, dispute_model, value_capture_model, autonomous_operating_loop. No token design yet. Thesis: " + json.dumps({"problem": pass_a or proposal.pitch, "onchain_review": pass_b}, default=str)
|
|
architecture = self._model_json(self.generation_model_hint, pass_c_prompt, purpose=ModelCapability.REASONING)
|
|
pass_d_prompt = "PASS D - NATIVE TOKEN UTILITY DESIGN. Derive the minimum native token utility only if it performs strong structural economic or security functions. Explore SECURITY_BOND, SLASHABLE_COLLATERAL, PROVIDER_ADMISSION, RESOURCE_ALLOCATION, MACHINE_ECONOMIC_IDENTITY, CONTRIBUTION_ACCOUNTING, SECURITY_BUDGET, PROTOCOL_FEE_ASSET, PROVIDER_REWARD, DEMAND_SIDE_PAYMENT, GOVERNANCE, ACCESS, TREASURY, OTHER. Payment/governance/rewards/access alone are insufficient. Return JSON with token_thesis, token_utility list, token_demand_loop list, value_capture, token_roles object where each role is REQUIRED/STRONGLY_USEFUL/USEFUL/OPTIONAL/UNNECESSARY, mechanism_fingerprint object, security_model, regulatory_risks. Architecture: " + json.dumps(architecture or pass_a or proposal.pitch, default=str)
|
|
token_design = self._model_json(self.generation_model_hint, pass_d_prompt, purpose=ModelCapability.REASONING)
|
|
usage["sol_protocol_architecture_requests" if self.generation_model_hint == "sol" else "qwen_requests"] += 1
|
|
usage["sol_token_design_requests" if self.generation_model_hint == "sol" else "qwen_requests"] += 1
|
|
if token_design:
|
|
crypto = {**crypto, "generation_passes": {**passes, "pass_c_protocol_architecture": architecture, "pass_d_token_utility": token_design}, "token_design_completed": True, "token_roles": token_design.get("token_roles", {}), "mechanism_fingerprint": token_design.get("mechanism_fingerprint", {})}
|
|
proposal.metadata = {**proposal.metadata, "crypto": crypto}
|
|
proposal.pitch = {**proposal.pitch, "Token thesis": str(token_design.get("token_thesis", proposal.pitch.get("Token thesis", ""))), "Token demand loop": self._as_list(token_design.get("token_demand_loop") or proposal.pitch.get("Token demand loop", []))}
|
|
proposal.save(update_fields=["metadata", "pitch", "updated_at"])
|
|
if hasattr(proposal, "protocol_thesis"):
|
|
proposal.protocol_thesis.token_thesis = str(token_design.get("token_thesis", proposal.protocol_thesis.token_thesis))
|
|
proposal.protocol_thesis.utility_categories = self._as_list(token_design.get("token_utility") or proposal.protocol_thesis.utility_categories)
|
|
proposal.protocol_thesis.save(update_fields=["token_thesis", "utility_categories", "updated_at"])
|
|
if hasattr(proposal, "token_demand_loop"):
|
|
proposal.token_demand_loop.loop = self._as_list(token_design.get("token_demand_loop") or proposal.token_demand_loop.loop)
|
|
proposal.token_demand_loop.save(update_fields=["loop", "updated_at"])
|
|
return usage
|
|
|
|
def _remove_weak_token_members(self, cohort: VentureCohort) -> dict[str, int]:
|
|
unnecessary = 0
|
|
optional = 0
|
|
essential = 0
|
|
strongly = 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, "routed_to_venture_studio": True, "crypto_failure_reason": f"token classification {assessment.classification}"}
|
|
proposal.save(update_fields=["status", "metadata", "updated_at"])
|
|
elif assessment.classification == TokenNecessityClassification.TOKEN_ESSENTIAL:
|
|
essential += 1
|
|
else:
|
|
strongly += 1
|
|
return {"token_unnecessary_rejections": cohort.metrics.get("token_unnecessary_rejections", 0) + unnecessary, "token_optional_route_to_saas": cohort.metrics.get("token_optional_route_to_saas", 0) + optional, "token_essential": cohort.metrics.get("token_essential", 0) + essential, "token_strongly_justified": cohort.metrics.get("token_strongly_justified", 0) + strongly, "serious_crypto_survivors": cohort.members.count(), "crypto_survivors": cohort.members.count()}
|
|
|
|
def light_market_research(self, cohort: VentureCohort) -> None:
|
|
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()
|
|
circuit_breaker = Counter()
|
|
avoided = 0
|
|
primary_sources = self.primary_source_seed_sources(proposal, query_plan)
|
|
all_sources.extend(primary_sources)
|
|
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:
|
|
if circuit_breaker and all(count >= (8 if depth == "deep" else 4) for count in circuit_breaker.values()):
|
|
avoided += 1
|
|
query_results.append({"category": category, "query": query, "status": "AVOIDED_DUE_TO_ENGINE_FAILURE", "result_count": 0, "tripped_engines": dict(circuit_breaker)})
|
|
continue
|
|
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:
|
|
status = self._engine_status(str(reason))
|
|
engine_health[status] += 1
|
|
if status in {"CAPTCHA", "RATE_LIMITED", "ACCESS_DENIED"}:
|
|
circuit_breaker[str(engine)] += 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=14 if depth == "deep" else 7) 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:
|
|
if url in seen:
|
|
rejected.append({**source, "quality": "reject", "rejection_reason": SOURCE_REJECTION_REASONS["DUPLICATE"], "relevance_score": 0})
|
|
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"], "source_tier": quality.get("source_tier", source.get("source_tier", "TIER_C")), "relevance_score": quality["score"], "acceptance_reason": quality["reason"]})
|
|
else:
|
|
rejected.append({**enriched, "quality": "weak", "rejection_reason": quality["reason"], "relevance_score": quality["score"]})
|
|
category_sources = {category: [source for source in accepted if source.get("category") == category] for category in CRYPTO_RESEARCH_CATEGORIES}
|
|
coverage_confidence = {category: self.category_coverage_confidence(sources) for category, sources in category_sources.items()}
|
|
coverage = {category: confidence in {"MEDIUM", "HIGH"} for category, confidence in coverage_confidence.items()}
|
|
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), "primary_source_count": len([s for s in accepted if s.get("source_tier") == "TIER_A"]), "coverage": coverage, "coverage_ratio": coverage_ratio, "coverage_confidence": coverage_confidence, "category_coverage_confidence": coverage_confidence, "source_provider_health": "SEARCH_INFRASTRUCTURE_FAILED" if search_infrastructure_failed else "SEARCH_WORKED", "search_engine_health": dict(engine_health), "engine_circuit_breaker_events": dict(circuit_breaker), "queries_avoided_due_to_engine_failure": avoided, "queries_avoided_due_to_circuit_breaker": avoided, "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([str(crypto.get("territory", "")), proposal.protocol_thesis.protocol_category, proposal.problem, proposal.target_customer]).replace("_", " ")
|
|
seed = seed.replace(proposal.title, "")
|
|
seed = re.sub(r"\s+", " ", seed).strip()[:120]
|
|
terms = {
|
|
"USER_PAIN": [f"{seed} user pain", f"{proposal.target_customer} coordination failure"],
|
|
"EXISTING_PROTOCOLS": [f"{seed} protocol documentation", f"{seed} crypto protocol precedent"],
|
|
"FAILED_PRECEDENTS": [f"{seed} failed crypto project", f"{seed} postmortem"],
|
|
"ONCHAIN_ALTERNATIVES": [f"{seed} onchain escrow settlement attestation"],
|
|
"OFFCHAIN_ALTERNATIVES": [f"{seed} SaaS alternative", f"{seed} offchain alternative"],
|
|
"TOKEN_MODELS": [f"{seed} token model staking", f"{seed} tokenomics"],
|
|
"COLLATERAL_MODELS": [f"{seed} collateral model stablecoin ETH staking"],
|
|
"STAKING_SLASHING_PRECEDENTS": [f"{seed} staking slashing", 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"],
|
|
"SECURITY_FAILURES": [f"{seed} security incident", f"{seed} exploit postmortem"],
|
|
"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": [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 primary_source_seed_sources(self, proposal: CompanyProposal, query_plan: dict[str, list[str]]) -> list[dict[str, Any]]:
|
|
seeds = []
|
|
category_urls = {
|
|
"ONCHAIN_ALTERNATIVES": ["https://ethereum.org/en/developers/docs/", "https://eips.ethereum.org/"],
|
|
"TOKEN_MODELS": ["https://ethereum.org/en/developers/docs/standards/tokens/"],
|
|
"COLLATERAL_MODELS": ["https://ethereum.org/en/defi/"],
|
|
"STAKING_SLASHING_PRECEDENTS": ["https://ethereum.org/en/staking/"],
|
|
"SECURITY_FAILURES": ["https://immunefi.com/blog/", "https://github.com/pcaversaccio/reentrancy-attacks"],
|
|
"LEGAL_REGULATORY": ["https://www.sec.gov/newsroom", "https://www.cftc.gov/PressRoom/PressReleases"],
|
|
}
|
|
for category in query_plan:
|
|
for url in category_urls.get(category, [])[:2]:
|
|
seeds.append({"type": "primary_seed", "source": "primary_source_seed", "source_tier": "TIER_A", "url": url, "title": url, "category": category, "summary": "Primary-source seed for crypto protocol research.", "fallback_evidence": False})
|
|
return seeds
|
|
|
|
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()
|
|
category = str(source.get("category", ""))
|
|
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": SOURCE_REJECTION_REASONS["SEARCH_NOISE"]}
|
|
source_tier = self.source_tier(url, text)
|
|
preferred = ["docs", "github", "whitepaper", "paper", "postmortem", "audit", "security", "research", "protocol", "token", "staking", "slashing", "stablecoin", "collateral", "attestation", "oracle", "eip"]
|
|
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)
|
|
category_terms = set(re.findall(r"[a-z0-9]{5,}", category.lower().replace("_", " ")))
|
|
category_overlap = len(category_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.025 + category_overlap * 0.12 + preferred_hits * 0.08 + (0.18 if source_tier == "TIER_A" else 0.08 if source_tier == "TIER_B" else 0)), 2)
|
|
if category and category_overlap == 0 and source_tier == "TIER_C" and score < 0.45:
|
|
return {"accepted": False, "score": score, "reason": SOURCE_REJECTION_REASONS["WRONG_CATEGORY"]}
|
|
if score >= 0.45:
|
|
return {"accepted": True, "quality": "strong", "source_tier": source_tier, "score": score, "reason": "strong category-relevant crypto primary/technical evidence"}
|
|
if score >= 0.28 and source_tier in {"TIER_A", "TIER_B"}:
|
|
return {"accepted": True, "quality": "medium", "source_tier": source_tier, "score": score, "reason": "medium authority category evidence"}
|
|
if source_tier == "TIER_C":
|
|
return {"accepted": False, "score": score, "reason": SOURCE_REJECTION_REASONS["LOW_AUTHORITY"]}
|
|
return {"accepted": False, "score": score, "reason": SOURCE_REJECTION_REASONS["IRRELEVANT"]}
|
|
|
|
def source_tier(self, url: str, text: str) -> str:
|
|
if any(domain in url for domain in PRIMARY_SOURCE_DOMAINS) or any(marker in url for marker in ["/docs", "docs.", "whitepaper", "forum.", "research."]):
|
|
return "TIER_A"
|
|
if any(marker in text or marker in url for marker in ["audit", "postmortem", "technical research", "security research", "protocol research"]):
|
|
return "TIER_B"
|
|
return "TIER_C"
|
|
|
|
def category_coverage_confidence(self, sources: list[dict[str, Any]]) -> str:
|
|
strong = len([source for source in sources if source.get("quality") == "strong"])
|
|
medium_domains = {self._domain(str(source.get("url", ""))) for source in sources if source.get("quality") == "medium"}
|
|
if strong >= 2:
|
|
return "HIGH"
|
|
if strong >= 1 or len(medium_domains) >= 2:
|
|
return "MEDIUM"
|
|
if sources:
|
|
return "LOW"
|
|
return "NONE"
|
|
|
|
def _domain(self, url: str) -> str:
|
|
return re.sub(r"^https?://(www\.)?", "", url.lower()).split("/")[0]
|
|
|
|
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 = []
|
|
for member in cohort.members.select_related("proposal"):
|
|
token = getattr(member.proposal, "token_utility_assessment", None)
|
|
if token and token.classification in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED}:
|
|
assessments.append(self.red_team_proposal(member.proposal))
|
|
counts = Counter(flag for item in assessments for flag in item.flags)
|
|
cohort.metrics = {**cohort.metrics, "token_red_team_failures": dict(counts), "token_red_team_count": len(assessments), "sol_token_red_team_requests": len(assessments) if self.final_ic_model_hint == "sol" else 0}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
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}})
|
|
model_usage = Counter(cohort.metrics.get("model_usage", {}))
|
|
model_usage["sol_token_utility_judge_requests"] = max(model_usage.get("sol_token_utility_judge_requests", 0), cohort.metrics.get("sol_token_utility_judge_requests", 0))
|
|
model_usage["sol_final_ic_requests"] += len(rows) if self.final_ic_model_hint == "sol" else 0
|
|
model_usage["qwen_requests"] += 0
|
|
cohort.metrics = {**cohort.metrics, "crypto_ranked_count": len(rows), "crypto_top_3_count": len(top_3), "autonomous_crypto_survivors": len([row for row in rows if row.get("crypto_autonomy_class") == CryptoAutonomyClass.AUTONOMOUS_CRYPTO]), "assisted_high_potential": len([row for row in rows if row.get("crypto_ic_decision") == CryptoICDecisionType.ASSISTED_CRYPTO_HIGH_POTENTIAL]), "finalists": len(top_3), "model_usage": dict(model_usage), "sol_token_utility_judge_requests": model_usage.get("sol_token_utility_judge_requests", 0), "sol_final_ic_requests": model_usage.get("sol_final_ic_requests", 0), "qwen_requests": model_usage.get("qwen_requests", 0)}
|
|
cohort.status = "CRYPTO_IC_COMPLETE"
|
|
cohort.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)
|
|
routed_to_saas = [{"company": proposal.title, "token_classification": getattr(proposal, "token_utility_assessment", None).classification if hasattr(proposal, "token_utility_assessment") else None, "product_thesis": proposal.description, "reason": "Failed onchain/token counterfactual but may be valuable as pre-token/SaaS product."} for proposal in cohort.mandate.company_proposals.filter(status=CompanyProposalStatus.REJECTED, metadata__routed_to_saas=True).order_by("created_at")[:10]]
|
|
comparison = {"v02": cohort.metadata.get("v02_baseline", {"raw": 30, "crypto_survivors": 2, "autonomous_survivors": 0, "finalists": 0}), "v03": cohort.metadata.get("v03_baseline", {}), "v031": {"raw": cohort.metrics.get("raw_generated", 0), "onchain_passed": cohort.metrics.get("onchain_passed", 0), "onchain_rejected": cohort.metrics.get("onchain_rejected", 0), "token_strongly_justified": cohort.metrics.get("token_strongly_justified", 0), "token_essential": cohort.metrics.get("token_essential", 0), "crypto_survivors": cohort.members.count(), "autonomous_survivors": cohort.metrics.get("autonomous_crypto_survivors", 0), "assisted_high_potential": cohort.metrics.get("assisted_high_potential", 0), "finalists": len(review.top_3), "average_research_quality": research_health.get("average_coverage", 0), "idea_diversity": review.concentration}}
|
|
security_summary = {"guard_reviews": cohort.metrics.get("protocol_security_gate_passed", 0) + cohort.metrics.get("protocol_security_gate_blocked", 0), "pass": cohort.metrics.get("protocol_security_gate_passed", 0), "revise_or_pass": cohort.metrics.get("protocol_security_gate_passed", 0), "block": cohort.metrics.get("protocol_security_gate_blocked", 0)}
|
|
content = {"title": "CRYPTO VENTURE COHORT V0.3.1", "cohort_id": cohort.cohort_id, "graph_run": str(cohort.graph_run_id or ""), "runtime": cohort.metrics, "raw_protocols_generated": cohort.metrics.get("raw_generated", 0), "raw_sol_theses": cohort.metrics.get("raw_sol_theses", 0), "onchain_rejected": cohort.metrics.get("onchain_rejected", 0), "onchain_passed": cohort.metrics.get("onchain_passed", 0), "token_design_attempts": cohort.metrics.get("token_design_attempts", 0), "token_design_skipped": cohort.metrics.get("token_design_skipped_due_to_onchain_rejection", 0), "token_unnecessary": cohort.metrics.get("token_unnecessary_rejections", 0), "token_optional_routed_saas": cohort.metrics.get("token_optional_route_to_saas", 0), "token_strongly_justified": cohort.metrics.get("token_strongly_justified", 0), "token_essential": cohort.metrics.get("token_essential", 0), "duplicates": cohort.metrics.get("duplicate_rejections", 0), "crypto_survivors": cohort.members.count(), "autonomous_crypto_survivors": cohort.metrics.get("autonomous_crypto_survivors", 0), "assisted_high_potential": cohort.metrics.get("assisted_high_potential", 0), "security_blocked": cohort.metrics.get("protocol_security_gate_blocked", 0), "finalists": len(review.top_3), "generation_attempts": cohort.metrics.get("generation_attempts", 0), "model_usage": cohort.metrics.get("model_usage", {}), "research_summary": research_health, "security_summary": security_summary, "ranking": rows, "top_3": review.top_3, "routed_to_saas": routed_to_saas, "v02_v03_v031_comparison": comparison, "v02_comparison": comparison, "token_red_team_failures": cohort.metrics.get("token_red_team_failures", {}), "token_utility_distribution": dict(Counter(util for row in rows for util in row.get("token_utility", []))), "crypto_thesis_saturation": review.concentration, "capability_gaps": review.capability_demand, "research_health": research_health, "system_metrics": cohort.metrics, "stop_conditions": {"token_sale": False, "nft_sale": False, "founding_membership_sale": False, "fundraising": False, "mainnet_issuance": False, "user_contact": False, "real_spend": 0}}
|
|
artifact = self._artifact(None, cohort.mandate, "CRYPTO_VENTURE_COHORT_REPORT", "Crypto Venture Cohort V0.3.1 Report", content, self._readable_report(content), "crypto_portfolio_ic", graph_run=cohort.graph_run)
|
|
cohort.status = "COMPLETE"
|
|
cohort.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
|
|
primary = 0
|
|
avoided = 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)
|
|
primary += int(research.get("primary_source_count", 0) or 0)
|
|
avoided += int(research.get("queries_avoided_due_to_circuit_breaker", 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, "primary_sources_accepted": primary, "queries_avoided_due_to_circuit_breaker": avoided, "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._design_text(proposal)
|
|
utilities = [u for u in crypto.get("token_utility", []) if isinstance(u, str)] or self._as_list(getattr(proposal, "protocol_thesis", None).utility_categories if hasattr(proposal, "protocol_thesis") else []) or self._infer_utilities(text)
|
|
roles = self.token_role_decomposition(proposal)
|
|
utility_profile = self.token_utility_profile(proposal, roles)
|
|
strong_count = sum(1 for utility in utilities if utility.lower() in ALLOWED_TOKEN_UTILITIES or any(term in utility.lower() for term in ["stake", "slash", "collateral", "fee", "marketplace", "attestation", "compute", "machine", "security"]))
|
|
strong_roles = [role for role, value in roles.roles.items() if value in {"REQUIRED", "STRONGLY_USEFUL"}]
|
|
structural_roles = [role for role in strong_roles if role in STRUCTURAL_TOKEN_ROLES]
|
|
required_roles = len(strong_roles)
|
|
weak = any(term in text for term in ["meme", "speculative", "community token", "marketing token", "community marketing", "governance token only", "token gated subscription"])
|
|
weak_only = bool(strong_roles) and not structural_roles and all(role in WEAK_TOKEN_ROLES for role in strong_roles)
|
|
payment_only = strong_roles == ["DEMAND_SIDE_PAYMENT"] or roles.roles.get("PROTOCOL_FEE_ASSET") in {"REQUIRED", "STRONGLY_USEFUL"} and len(strong_roles) <= 1
|
|
score = min(100, 35 + strong_count * 4 + len(strong_roles) * 8 + len(structural_roles) * 13 + (12 if "slash" in text or "slashing" in text else 0) + (8 if "provider" in text else 0) - (35 if payment_only or weak_only else 0) - (35 if weak else 0) - int(utility_profile.get("SPECULATION_DEPENDENCE", 0) * 0.25))
|
|
economic_model = getattr(proposal, "token_economic_model", None)
|
|
model_payload = {"unit_of_service": economic_model.unit_of_service, "payment_asset": economic_model.payment_asset, "stake_requirement": economic_model.stake_requirement, "collateral_requirement": economic_model.collateral_requirement} if economic_model else {}
|
|
sol_review = self._sol_json("CRYPTO_TOKEN_UTILITY_JUDGE V0.3.1. Independent from ideation, onchain review, and token design. Challenge the token; do not merely summarize it. A native token need not be mathematically impossible to substitute, but benefits must outweigh costs. Evaluate security quality, attack economics, slashing effectiveness, provider skin-in-the-game, capital efficiency, protocol-specific risk pricing, permissionless supply formation, participant coordination, resource allocation, reputation portability, value capture, network effects, bootstrapping economics, governance, and composability versus USDC/ETH/external collateral/non-transferable attestations. Return JSON with token_classification TOKEN_ESSENTIAL/TOKEN_STRONGLY_JUSTIFIED/TOKEN_OPTIONAL/TOKEN_UNNECESSARY; score 0-100; utility_scores object; strong_structural_roles list; weak_roles list; counterfactual_analysis; material_improvements_over_USDC_ETH list; native_asset_costs_and_risks list; argument_for_native_token; argument_against_native_token; final_rationale; native_token_removed_breaks list; native_token_removed_degradation NONE/MINOR/MODERATE/MATERIAL/CRITICAL; native_token_removed_explanation. Product/protocol/token/economic model: " + json.dumps({"pitch": proposal.pitch, "crypto": crypto, "roles": roles.roles, "utility_profile": utility_profile, "economic_model": model_payload}, default=str))
|
|
sol_structural_roles = [str(role) for role in self._as_list((sol_review or {}).get("strong_structural_roles")) if str(role) in STRUCTURAL_TOKEN_ROLES]
|
|
sol_scores = (sol_review or {}).get("utility_scores", {}) if isinstance((sol_review or {}).get("utility_scores", {}), dict) else {}
|
|
if sol_structural_roles:
|
|
structural_roles = sorted(set(structural_roles) | set(sol_structural_roles))
|
|
if sol_scores:
|
|
utility_profile = {**utility_profile, **{key: int(float(value)) for key, value in sol_scores.items() if key in TOKEN_UTILITY_SCORE_KEYS}}
|
|
if sol_review:
|
|
score = float(sol_review.get("score", score))
|
|
sol_classification = str(sol_review.get("token_classification") or sol_review.get("classification", ""))
|
|
else:
|
|
sol_classification = ""
|
|
if len(strong_roles) >= 2 and structural_roles and score >= 85:
|
|
classification = TokenNecessityClassification.TOKEN_ESSENTIAL
|
|
elif len(strong_roles) >= 2 and structural_roles and score >= 75:
|
|
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
|
|
degradation = str((sol_review or {}).get("native_token_removed_degradation", "MATERIAL" if structural_roles else "MINOR"))
|
|
no_native_claim = any(phrase in text for phrase in ["native token not required", "no native token required", "no proprietary token required"])
|
|
has_explicit_improvement = bool(self._as_list((sol_review or {}).get("material_improvements_over_USDC_ETH")) or self._as_list((sol_review or {}).get("native_token_removed_breaks")) or (sol_review or {}).get("argument_for_native_token"))
|
|
if weak:
|
|
classification = TokenNecessityClassification.TOKEN_UNNECESSARY
|
|
score = min(score, 35)
|
|
elif no_native_claim and not has_explicit_improvement:
|
|
classification = TokenNecessityClassification.TOKEN_UNNECESSARY
|
|
score = min(score, 35)
|
|
elif payment_only or weak_only or not structural_roles:
|
|
classification = TokenNecessityClassification.TOKEN_OPTIONAL if score >= 45 else TokenNecessityClassification.TOKEN_UNNECESSARY
|
|
score = min(score, 62)
|
|
elif classification in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED} and not has_explicit_improvement:
|
|
classification = TokenNecessityClassification.TOKEN_OPTIONAL
|
|
score = min(score, 62)
|
|
if (len(strong_roles) < 2 or not structural_roles or degradation in {"NONE", "MINOR"}) and classification in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED}:
|
|
classification = TokenNecessityClassification.TOKEN_OPTIONAL
|
|
score = min(score, 62)
|
|
proposal.metadata = {**proposal.metadata, "crypto": {**crypto, "token_utility_judge_model": self.final_ic_model_hint, "token_utility_judge_used": bool(sol_review), "token_utility_profile": utility_profile, "strong_structural_roles": structural_roles, "strong_token_roles": strong_roles}}
|
|
proposal.save(update_fields=["metadata", "updated_at"])
|
|
return TokenUtilityAssessment.objects.update_or_create(proposal=proposal, defaults={"classification": classification, "token_necessity_score": score, "utility_categories": utilities, "fiat_or_database_substitution": str((sol_review or {}).get("counterfactual_analysis", "USDC/ETH/external collateral may preserve basic operation but must be compared against protocol-specific security and coordination improvements.")), "rationale": str((sol_review or {}).get("final_rationale") or (sol_review or {}).get("rationale") or "Token score derives from structural utility, not native payment currency alone."), "metadata": {"weak_utility_terms": INSUFFICIENT_TOKEN_UTILITIES, "sol_counterfactual_review": sol_review or {}, "sol_token_utility_judge": sol_review or {}, "independent_token_utility_judge": "CRYPTO_TOKEN_UTILITY_JUDGE", "argument_for_native_token": (sol_review or {}).get("argument_for_native_token", ""), "argument_against_native_token": (sol_review or {}).get("argument_against_native_token", ""), "utility_scores": utility_profile, "strong_structural_roles": structural_roles, "strong_token_roles": strong_roles, "weak_roles": (sol_review or {}).get("weak_roles", []), "counterfactual_analysis": (sol_review or {}).get("counterfactual_analysis", ""), "external_collateral_counterfactual": (sol_review or {}).get("external_collateral_counterfactual") or (sol_review or {}).get("model_c_external_collateral", {}), "material_improvements_over_USDC_ETH": self._as_list((sol_review or {}).get("material_improvements_over_USDC_ETH")), "native_asset_costs_and_risks": self._as_list((sol_review or {}).get("native_asset_costs_and_risks")), "required_roles": roles.required_roles, "payment_only_penalty": payment_only, "native_token_removed_breaks": self._as_list((sol_review or {}).get("native_token_removed_breaks")), "native_token_removed_degradation": degradation, "native_token_removed_explanation": (sol_review or {}).get("native_token_removed_explanation", "")}})[0]
|
|
|
|
def token_role_decomposition(self, proposal: CompanyProposal) -> TokenRoleDecomposition:
|
|
text = self._design_text(proposal)
|
|
explicit_roles = proposal.metadata.get("crypto", {}).get("token_roles", {}) if isinstance(proposal.metadata, dict) else {}
|
|
roles = {}
|
|
role_terms = {
|
|
"SECURITY_BOND": ["bond", "security budget"],
|
|
"SLASHABLE_COLLATERAL": ["slash", "slashing", "collateral"],
|
|
"PROVIDER_ADMISSION": ["provider admission", "permissionless provider", "admission"],
|
|
"RESOURCE_ALLOCATION": ["resource allocation", "scarce", "quota", "bandwidth", "storage", "compute slot"],
|
|
"MACHINE_ECONOMIC_IDENTITY": ["machine economic", "machine identity", "agent identity", "reputation"],
|
|
"CONTRIBUTION_ACCOUNTING": ["contribution", "contributor", "accounting"],
|
|
"SECURITY_BUDGET": ["security budget"],
|
|
"PROTOCOL_FEE_ASSET": ["protocol fee", "fee asset"],
|
|
"PROVIDER_REWARD": ["provider reward", "reward"],
|
|
"DEMAND_SIDE_PAYMENT": ["payment", "settlement"],
|
|
"GOVERNANCE": ["governance", "vote"],
|
|
"ACCESS": ["access"],
|
|
"TREASURY": ["treasury"],
|
|
"OTHER": [],
|
|
}
|
|
for role, terms in role_terms.items():
|
|
explicit = str(explicit_roles.get(role, "")).upper() if isinstance(explicit_roles, dict) else ""
|
|
if explicit in {"REQUIRED", "STRONGLY_USEFUL", "USEFUL", "OPTIONAL", "UNNECESSARY"}:
|
|
value = explicit
|
|
elif explicit_roles:
|
|
value = TokenRoleRequirement.UNNECESSARY
|
|
elif role == "GOVERNANCE" and any(term in text for term in terms) and not any(term in text for term in ["parameter", "slashing", "treasury"]):
|
|
value = TokenRoleRequirement.OPTIONAL
|
|
elif any(term in text for term in terms):
|
|
value = "REQUIRED" if role in STRUCTURAL_TOKEN_ROLES else "USEFUL"
|
|
else:
|
|
value = TokenRoleRequirement.UNNECESSARY
|
|
roles[role] = value
|
|
required = [role for role, value in roles.items() if value in {"REQUIRED", "STRONGLY_USEFUL"}]
|
|
stablecoin = {"MODEL_A_native_payment_staking_governance": "baseline proposal", "MODEL_B_stablecoin_payment_native_bond": "preferred if payment utility is separable", "MODEL_C_stablecoin_payment_stablecoin_collateral": "valid if slashing/collateral does not need protocol-native exposure", "MODEL_D_onchain_no_proprietary_token": "valid if proprietary token adds no security/resource allocation advantage", "MODEL_E_centralized_saas_database": "routes to SaaS if verification/settlement/reputation do not degrade", "usdc_identical_payment_utility": roles.get("DEMAND_SIDE_PAYMENT") not in {"REQUIRED", "STRONGLY_USEFUL"}, "external_collateral_identical_security": "SLASHABLE_COLLATERAL" not in required and "SECURITY_BOND" not in required, "stable_collateral_identical_security": "SLASHABLE_COLLATERAL" not in required and "SECURITY_BOND" not in required}
|
|
outcome = "Native token removal materially degrades security/reputation if required roles remain." if required else "Native token removed changes little; route to SaaS or revise token model."
|
|
return TokenRoleDecomposition.objects.update_or_create(proposal=proposal, defaults={"roles": roles, "required_roles": required, "stablecoin_counterfactual": stablecoin, "native_token_removed_outcome": outcome})[0]
|
|
|
|
def token_utility_profile(self, proposal: CompanyProposal, roles: TokenRoleDecomposition) -> dict[str, int]:
|
|
text = self._design_text(proposal)
|
|
score = {key: 0 for key in TOKEN_UTILITY_SCORE_KEYS}
|
|
score["SECURITY_UTILITY"] = 85 if roles.roles.get("SECURITY_BOND") in {"REQUIRED", "STRONGLY_USEFUL"} else 35
|
|
score["COLLATERAL_UTILITY"] = 85 if roles.roles.get("SLASHABLE_COLLATERAL") in {"REQUIRED", "STRONGLY_USEFUL"} else 30
|
|
score["SLASHING_UTILITY"] = 82 if "slash" in text and roles.roles.get("SLASHABLE_COLLATERAL") in {"REQUIRED", "STRONGLY_USEFUL"} else 25
|
|
score["RESOURCE_ALLOCATION_UTILITY"] = 82 if roles.roles.get("RESOURCE_ALLOCATION") in {"REQUIRED", "STRONGLY_USEFUL"} else 25
|
|
score["MACHINE_ECONOMIC_UTILITY"] = 84 if roles.roles.get("MACHINE_ECONOMIC_IDENTITY") in {"REQUIRED", "STRONGLY_USEFUL"} else 25
|
|
score["COORDINATION_UTILITY"] = 75 if any(roles.roles.get(role) in {"REQUIRED", "STRONGLY_USEFUL"} for role in ["PROVIDER_ADMISSION", "CONTRIBUTION_ACCOUNTING", "RESOURCE_ALLOCATION"]) else 35
|
|
score["INCENTIVE_UTILITY"] = 70 if roles.roles.get("PROVIDER_REWARD") in {"REQUIRED", "STRONGLY_USEFUL", "USEFUL"} else 25
|
|
score["NETWORK_BOOTSTRAP_UTILITY"] = 75 if "bootstrap" in text or roles.roles.get("PROVIDER_ADMISSION") in {"REQUIRED", "STRONGLY_USEFUL"} else 30
|
|
score["VALUE_CAPTURE_UTILITY"] = 75 if "value" in text or roles.roles.get("PROTOCOL_FEE_ASSET") in {"REQUIRED", "STRONGLY_USEFUL", "USEFUL"} else 30
|
|
score["GOVERNANCE_UTILITY"] = 45 if roles.roles.get("GOVERNANCE") in {"REQUIRED", "STRONGLY_USEFUL", "USEFUL"} else 10
|
|
score["PAYMENT_UTILITY"] = 45 if roles.roles.get("DEMAND_SIDE_PAYMENT") in {"REQUIRED", "STRONGLY_USEFUL", "USEFUL"} else 10
|
|
score["SPECULATION_DEPENDENCE"] = 70 if any(self._positive_phrase_present(text, term) for term in ["price appreciation", "buyback", "yield", "apy", "speculat", "token price"]) else 5
|
|
return score
|
|
|
|
def assess_pretoken_monetization(self, proposal: CompanyProposal) -> PreTokenMonetizationAssessment:
|
|
text = self._design_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}
|
|
customer = proposal.target_customer[:180]
|
|
product = proposal.description[:220]
|
|
payment = proposal.pricing_hypothesis or "Stripe/stablecoin usage credits"
|
|
ladder = {"1000": f"Sell 10 x $100 paid beta/API-credit packages to {customer} for {product}; fulfill with hosted testnet/API access and automated reports.", "5000": f"Sell 20 x $250 monthly usage-credit packages via {payment}; buyers get measurable protocol simulations, SDK/API access, and evidence dashboards before any token.", "10000": f"Sell 20 x $500 subscription/service-credit plans; fulfillment is self-service onboarding, usage metering, testnet jobs, and downloadable verification evidence."}
|
|
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, service credits, 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._design_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, "conditions": ["missed settlement", "invalid proof", "fraudulent attestation"]}, "slash_amount": round(stake * 0.15, 2), "emission_schedule": {"v03": "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": {"v03": "none"}, "circulation_assumptions": {"v03_native_token_supply": 0}, "metadata": {"buyer": proposal.target_customer, "provider": "permissionless providers/operators serving the unit of service", "average_transaction": avg_fee, "protocol_fee": round(avg_fee * 0.2, 3), "provider_reward": round(avg_fee * 0.8, 3), "token_sources": ["earned service fees", "bond acquisition only after legal gate"], "provider_economics": {"gross_fee_share": 0.8, "slash_risk": 0.01, "stake_requirement": stake}}})[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")
|
|
if payload.get("generation_passes"):
|
|
proposal.metadata = {**proposal.metadata, "crypto": {**proposal.metadata["crypto"], "generation_passes": payload.get("generation_passes", {}), "problem_first_pass_a_had_no_token": True, "pre_token_product": payload.get("pre_token_product"), "pre_token_customer": payload.get("pre_token_customer"), "pre_token_value": payload.get("pre_token_value"), "pre_token_payment_method": payload.get("pre_token_payment_method"), "pre_token_revenue_model": payload.get("pre_token_revenue_model")}}
|
|
proposal.save(update_fields=["metadata", "updated_at"])
|
|
if payload.get("token_roles"):
|
|
proposal.metadata = {**proposal.metadata, "crypto": {**proposal.metadata["crypto"], "token_roles": payload.get("token_roles", {})}}
|
|
proposal.save(update_fields=["metadata", "updated_at"])
|
|
ProtocolThesis.objects.create(proposal=proposal, product_thesis=product, protocol_thesis=protocol, token_thesis=token, network_effect=proposal.metadata["crypto"]["network_effect"], bootstrap_plan=proposal.metadata["crypto"]["bootstrap_plan"], autonomous_operability="Artifex can build/testnet deploy/monitor only; mainnet and issuance stop at human legal gate.", utility_categories=proposal.metadata["crypto"]["token_utility"], protocol_category=territory)
|
|
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, Counter]:
|
|
usage = Counter()
|
|
if self.router is not None:
|
|
try:
|
|
pass_a_prompt = "PASS A - PROBLEM / NETWORK DESIGN. Generate exactly one protocol venture problem thesis as JSON with NO token fields and no token design. Include name, one_line_thesis, product_thesis, user, participants, supply_side, demand_side, coordination_failure, trust_problem, why_neutral_network_may_help, pre_token_product, pre_token_customer, pre_token_value, pre_token_payment_method, pre_token_revenue_model, revenue_1000_path, revenue_5000_path, revenue_10000_path, autonomous_operating_loop, bootstrap_plan, validation_experiment, major_risks, confidence. Do not include token_thesis, token_utility, tokenomics, governance token, rewards token, or staking token. Territory: " + territory + ". Exclude API Reliability Slashing Pool, VeriMesh, VeriChain, and simple staking/slashing agent marketplaces unless the core coordination mechanism is materially different."
|
|
pass_a = self._model_json(self.generation_model_hint, pass_a_prompt, purpose=ModelCapability.PLANNING)
|
|
usage["sol_generation_requests" if self.generation_model_hint == "sol" else "qwen_requests"] += 1
|
|
if pass_a and not any(key in pass_a for key in ["token_thesis", "token_utility", "tokenomics"]):
|
|
pass_b_prompt = "PASS B - ONCHAIN NECESSITY. Independently evaluate whether this coordination system materially improves when implemented onchain. Return JSON with onchain_necessity_score 0-100, why_onchain, offchain_substitute, pass_onchain boolean, reasons list. Consider trust minimization, neutral settlement, permissionless participation, verifiable state, escrow, programmable guarantees, censorship resistance, cross-organization coordination, and machine-to-machine interaction. Thesis: " + json.dumps(pass_a, default=str)
|
|
pass_b = self._model_json(self.generation_model_hint, pass_b_prompt, purpose=ModelCapability.REASONING)
|
|
usage["sol_onchain_judge_requests" if self.generation_model_hint == "sol" else "qwen_requests"] += 1
|
|
payload = {**pass_a, **pass_b, "token_thesis": "Native token not assessed yet; token utility design runs only after onchain qualification.", "token_utility": [], "token_demand_loop": [], "user_pain": pass_a.get("coordination_failure") or pass_a.get("trust_problem"), "protocol_thesis": pass_a.get("why_neutral_network_may_help") or pass_a.get("product_thesis"), "business_model": pass_a.get("pre_token_revenue_model"), "fee_model": pass_a.get("pre_token_payment_method"), "generation_passes": {"pass_a_problem_network": pass_a, "pass_b_onchain": pass_b}}
|
|
return payload, self.generation_model_hint, usage
|
|
except Exception:
|
|
pass
|
|
usage["fallback_count"] += 1
|
|
return self._fallback_payload(index, territory), "deterministic_crypto_fallback", usage
|
|
|
|
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": CRYPTO_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("TOKEN_NOT_REQUIRED")
|
|
if onchain.onchain_necessity_score < 70:
|
|
flags.append("REJECT_ONCHAIN_NOT_NEEDED")
|
|
if value.value_accrual_quality_score < 65:
|
|
flags.append("VALUE_CAPTURE_WEAK")
|
|
if value.tokenomics_sustainability_score < 65:
|
|
flags.append("UNSUSTAINABLE_EMISSIONS")
|
|
if policy.regulatory_manageability_score < 55:
|
|
flags.append("LEGAL_COMPLEXITY_OUTWEIGHS_UTILITY")
|
|
text = self._design_text(proposal)
|
|
if "speculat" in text or "apy" in text or "yield farm" in text:
|
|
flags.append("SPECULATION_DEPENDENT")
|
|
if "governance" in text and "slash" not in text and "fee" not in text:
|
|
flags.append("GOVERNANCE_ONLY_TOKEN")
|
|
if len([role for role, value in roles.roles.items() if value in {TokenRoleRequirement.REQUIRED, TokenRoleRequirement.USEFUL}]) >= 6:
|
|
flags.append("TOKEN_DOES_TOO_MUCH")
|
|
if roles.roles.get("DEMAND_SIDE_PAYMENT") in {"USEFUL", "OPTIONAL", "REQUIRED", "STRONGLY_USEFUL"} and not any(roles.roles.get(role) in {"REQUIRED", "STRONGLY_USEFUL"} for role in STRUCTURAL_TOKEN_ROLES):
|
|
flags.append("PAYMENT_ONLY_TOKEN")
|
|
flags.append("NATIVE_PAYMENT_NOT_REQUIRED")
|
|
if roles.stablecoin_counterfactual.get("stable_collateral_identical_security"):
|
|
flags.append("COLLATERAL_UTILITY_WEAK")
|
|
if any(term in text for term in ["buyback", "revenue share", "profit share"]):
|
|
flags.append("REVENUE_CLAIM_LANGUAGE")
|
|
if "buyback" in text:
|
|
flags.append("BUYBACK_DEPENDENCY")
|
|
if "yield" in text or "apy" in text:
|
|
flags.append("YIELD_DEPENDENCY")
|
|
if "manual dispute" in text or "human court" in text:
|
|
flags.append("SLASHING_DEPENDS_ON_HUMAN_JUDGMENT")
|
|
if "central operator" in text:
|
|
flags.append("CENTRAL_OPERATOR_CONTRADICTION")
|
|
if "subjective" in text and "slashing" in text:
|
|
flags.append("SLASHING_NOT_OBJECTIVE")
|
|
sol_review = self._sol_json("Independent Token Red Team V0.3.1. Return JSON with flags list using only TOKEN_DOES_TOO_MUCH, PAYMENT_ONLY_TOKEN, GOVERNANCE_ONLY_TOKEN, REWARD_ONLY_TOKEN, NATIVE_PAYMENT_NOT_REQUIRED, COLLATERAL_UTILITY_WEAK, SLASHING_NOT_OBJECTIVE, SLASHING_DEPENDS_ON_HUMAN_JUDGMENT, TOKEN_SECURITY_NOT_LINKED_TO_PROTOCOL_RISK, SPECULATION_DEPENDENT, REVENUE_CLAIM_LANGUAGE, BUYBACK_DEPENDENCY, YIELD_DEPENDENCY, UNSUSTAINABLE_EMISSIONS, MERCENARY_PROVIDER_INCENTIVES, RESOURCE_ALLOCATION_ARTIFICIAL, MACHINE_IDENTITY_NOT_PORTABLE, VALUE_CAPTURE_WEAK, BOOTSTRAP_REQUIRES_TOKEN_PRICE_GROWTH, CENTRAL_OPERATOR_CONTRADICTION, LEGAL_COMPLEXITY_OUTWEIGHS_UTILITY; severity LOW/MEDIUM/HIGH; critique. Proposal: " + json.dumps(self.crypto_score_context(proposal), default=str))
|
|
if sol_review:
|
|
for flag in self._as_list(sol_review.get("flags")):
|
|
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))
|
|
generation_passes = proposal.metadata.get("crypto", {}).get("generation_passes", {})
|
|
pass_b = generation_passes.get("pass_b_onchain", {}) if isinstance(generation_passes, dict) else {}
|
|
if pass_b:
|
|
score = float(pass_b.get("onchain_necessity_score", score))
|
|
return OnchainNecessityAssessment.objects.update_or_create(proposal=proposal, defaults={"onchain_necessity_score": score, "reasons": self._as_list(pass_b.get("reasons")) or [term for term in terms if term in text], "offchain_substitute": str(pass_b.get("offchain_substitute") or "Private SaaS/database is acceptable only if escrow, slashing, public reputation, and neutral settlement are not material."), "rationale": str(pass_b.get("why_onchain") or "Scores onchain necessity from trust-minimized coordination, settlement, staking/slashing, attestations, and multi-party neutrality."), "metadata": {"generation_pass_b": pass_b}})[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": 0.8, "TOKEN_PRICE_UP_10X": 1.2, "PROVIDER_CHURN_50": 0.7, "USER_GROWTH_10X": 10.0, "HIGH_SLASH_RATE": 1.0, "ZERO_EMISSIONS": 1.0, "STABLECOIN_PAYMENT": 1.0, "ETH_COLLATERAL": 1.0, "USDC_COLLATERAL": 1.0, "NATIVE_TOKEN_REMOVED": 1.0}
|
|
for name, multiplier in scenario_multipliers.items():
|
|
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_EMISSIONS"} else round(max(0, tx * 0.01), 2)
|
|
if name == "PROVIDER_CHURN_50":
|
|
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 in {"STABLECOIN_PAYMENT", "ETH_COLLATERAL", "USDC_COLLATERAL"}
|
|
degradation_dimensions = {"SECURITY_DEGRADATION": "SECURITY_BOND" in roles.required_roles or "SLASHABLE_COLLATERAL" in roles.required_roles, "COORDINATION_DEGRADATION": "PROVIDER_ADMISSION" in roles.required_roles, "RESOURCE_ALLOCATION_DEGRADATION": "RESOURCE_ALLOCATION" in roles.required_roles, "BOOTSTRAP_DEGRADATION": "DECENTRALIZED_SUPPLY_COORDINATION" in roles.required_roles, "VALUE_CAPTURE_DEGRADATION": "PROTOCOL_FEE_ASSET" in roles.required_roles, "NETWORK_EFFECT_DEGRADATION": bool(roles.required_roles), "MACHINE_ECONOMIC_DEGRADATION": "MACHINE_ECONOMIC_IDENTITY" in roles.required_roles}
|
|
protocol_degrades = native_removed and any(degradation_dimensions.values())
|
|
scenarios[name] = {"users": users, "transactions": tx, "fees": fees, "token_demand": 0 if native_removed else fees + staking * 0.01, "provider_supply": provider_supply, "staking": 0 if native_removed else staking, "emissions": emissions, "treasury": round(1000 + fees * model.treasury_share - emissions * 0.1, 2), "circulating_supply": 0 if native_removed else 1_000_000 + emissions, "token_velocity": 0 if native_removed else round(tx / max(1, fees + staking * 0.01), 2), "reward_coverage": round(fees / max(1, emissions), 2), "network_security_budget": 0 if native_removed else staking, "slash_events": round(tx * slash_rate, 2), "stablecoin_payment_works": stablecoin_substitute or roles.roles.get("DEMAND_SIDE_PAYMENT") not in {"REQUIRED", "STRONGLY_USEFUL"}, "external_collateral_works": name in {"ETH_COLLATERAL", "USDC_COLLATERAL"} and roles.stablecoin_counterfactual.get("external_collateral_identical_security"), "protocol_degrades_without_native_token": protocol_degrades, "native_token_removed_degradation_dimensions": degradation_dimensions if native_removed else {}}
|
|
summary = {"token_price_appreciation_primary_success_variable": False, "subsidy_removed_survives": scenarios["NO_SUBSIDY"]["reward_coverage"] >= 1.0, "native_token_removed_degrades_protocol": scenarios["NATIVE_TOKEN_REMOVED"]["protocol_degrades_without_native_token"], "native_token_removed_degradation_dimensions": scenarios["NATIVE_TOKEN_REMOVED"]["native_token_removed_degradation_dimensions"], "stablecoin_payment_substitute_works": scenarios["STABLECOIN_PAYMENT"]["stablecoin_payment_works"], "eth_collateral_substitute_works": scenarios["ETH_COLLATERAL"]["external_collateral_works"], "usdc_collateral_substitute_works": scenarios["USDC_COLLATERAL"]["external_collateral_works"], "stress_notes": ["Price appreciation is not a success variable.", "Native-token-removed asks whether protocol becomes materially worse, not literally impossible."]}
|
|
return TokenomicsSimulation.objects.update_or_create(proposal=proposal, defaults={"scenarios": scenarios, "summary": summary})[0]
|
|
|
|
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)
|
|
utility_profile = token.metadata.get("utility_scores", proposal.metadata.get("crypto", {}).get("token_utility_profile", {})) if isinstance(token.metadata, dict) else {}
|
|
text = self._proposal_text(proposal)
|
|
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
|
|
counterfactual = token.metadata.get("sol_token_utility_judge", {}) if isinstance(token.metadata, dict) else {}
|
|
return {"proposal_id": str(proposal.id), "company": proposal.title, "one_line_thesis": proposal.pitch.get("One-line thesis", ""), "product_thesis": proposal.protocol_thesis.product_thesis, "protocol_thesis": proposal.protocol_thesis.protocol_thesis, "user": proposal.target_customer, "pre_token_business": pretoken.pre_token_business, "pre_token_monetization": pretoken.mode, "pre_token_monetization_potential": pretoken.pre_token_monetization_potential, "pre_token_revenue_ladder": pretoken.revenue_ladder, "protocol_economy": proposal.protocol_thesis.protocol_thesis, "token_economy": proposal.protocol_thesis.token_thesis, "token_thesis": proposal.protocol_thesis.token_thesis, "why_onchain": proposal.onchain_assessment.rationale, "why_token": token.rationale, "token_role_decomposition": roles.roles, "token_utility_scores": utility_profile, "strong_token_roles": token.metadata.get("strong_token_roles", []), "strong_structural_roles": token.metadata.get("strong_structural_roles", []), "argument_for_native_token": token.metadata.get("argument_for_native_token", ""), "argument_against_native_token": token.metadata.get("argument_against_native_token", ""), "stablecoin_counterfactual": roles.stablecoin_counterfactual, "external_collateral_counterfactual": token.metadata.get("external_collateral_counterfactual", counterfactual.get("model_c_external_collateral", {})), "native_token_removed_breaks": token.metadata.get("native_token_removed_breaks", []), "native_token_removed_degradation": token.metadata.get("native_token_removed_degradation", ""), "native_token_removed_explanation": token.metadata.get("native_token_removed_explanation", ""), "native_token_removed_outcome": roles.native_token_removed_outcome, "native_token_removed_degrades_protocol": sim.summary.get("native_token_removed_degrades_protocol", False), "token_utility": token.utility_categories, "token_necessity_classification": token.classification, "token_demand_loop": proposal.token_demand_loop.loop, "value_capture": proposal.value_capture.value_accrual, "network_effect": proposal.protocol_thesis.network_effect, "bootstrap_plan": proposal.protocol_thesis.bootstrap_plan, "autonomous_operability": autonomy.autonomous_operability_score, "crypto_autonomy_class": autonomy_class, "regulatory_manageability": policy.regulatory_manageability_score, "security_assessment": security.decision if security else "NOT_ASSESSED", "security_risk": component_scores["SECURITY_MODEL_QUALITY"], "guard_verdict": security.decision if security else "NOT_ASSESSED", "validation_experiment": proposal.validation_plan, "component_scores": component_scores, "value_accrual_quality": value.value_accrual_quality_score, "tokenomics_sustainability": value.tokenomics_sustainability_score, "simulation_summary": sim.summary, "token_red_team_flags": red.flags, "crypto_ic_decision": decision, "crypto_ic_score": score, "research": proposal.metadata.get("research", {}), "research_confidence": proposal.metadata.get("research", {}).get("research_status", "RESEARCH_INSUFFICIENT"), "legal_review_required": policy.legal_review_required}
|
|
|
|
def assess_autonomy(self, proposal: CompanyProposal) -> AutonomousOperabilityAssessment:
|
|
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.REJECT_TOKEN_UTILITY_WEAK
|
|
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"] >= 85 and scores["ONCHAIN_NECESSITY"] >= 80 and scores["REAL_USAGE_DEMAND"] >= 75 and 55 <= scores["AUTONOMOUS_OPERABILITY"] < 70:
|
|
return CryptoICDecisionType.ASSISTED_CRYPTO_HIGH_POTENTIAL
|
|
if scores["TOKEN_NECESSITY"] >= 82 and scores["ONCHAIN_NECESSITY"] >= 75:
|
|
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 V0.3.1. Consider problem thesis, protocol thesis, token utility profile, independent token judge, Token Red Team, Guard assessment, token economic model, simulation, primary-source evidence, autonomy analysis, pre-token business, and uncertainty. Return JSON with optional score_adjustment -20..20, decision from TESTNET_PILOT, PROTOCOL_VALIDATE, REVISE_TOKEN_MODEL, ASSISTED_CRYPTO_HIGH_POTENTIAL, WATCHLIST, ROUTE_TO_VENTURE_STUDIO, REJECT_TOKEN_NOT_NEEDED, REJECT_TOKEN_UTILITY_WEAK, REJECT_ONCHAIN_NOT_NEEDED, REJECT_SECURITY_MODEL, REJECT_ECONOMIC_MODEL, REJECT_AUTONOMY, REJECT_NO_DEMAND, and rationale. Do not promote weak token ideas or fill finalist slots. Row: " + json.dumps(row, default=str))
|
|
if not review:
|
|
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"]
|
|
structural_roles = set(row.get("strong_structural_roles", []))
|
|
return row["token_necessity_classification"] in {TokenNecessityClassification.TOKEN_ESSENTIAL, TokenNecessityClassification.TOKEN_STRONGLY_JUSTIFIED} and len(row.get("strong_token_roles", [])) >= 2 and structural_roles & STRUCTURAL_TOKEN_ROLES and scores["TOKEN_NECESSITY"] >= 75 and scores["REAL_USAGE_DEMAND"] >= 65 and scores["ONCHAIN_NECESSITY"] >= 70 and scores["VALUE_ACCRUAL_QUALITY"] >= 65 and scores["TOKENOMICS_SUSTAINABILITY"] >= 65 and scores["AUTONOMOUS_OPERABILITY"] >= 70 and row.get("security_assessment") != ProtocolSecurityDecision.BLOCK_TESTNET and row.get("native_token_removed_degrades_protocol") is True
|
|
|
|
def _crypto_concentration(self, rows: list[dict[str, Any]]) -> dict[str, Any]:
|
|
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 _design_text(self, proposal: CompanyProposal) -> str:
|
|
parts = [proposal.title, proposal.description, proposal.problem, proposal.target_customer, proposal.proposed_solution, proposal.business_model, proposal.pricing_hypothesis, proposal.acquisition_strategy, proposal.validation_plan]
|
|
if hasattr(proposal, "protocol_thesis"):
|
|
parts.extend([proposal.protocol_thesis.product_thesis, proposal.protocol_thesis.protocol_thesis, proposal.protocol_thesis.token_thesis, proposal.protocol_thesis.network_effect, proposal.protocol_thesis.bootstrap_plan, json.dumps(proposal.protocol_thesis.utility_categories, default=str)])
|
|
if hasattr(proposal, "token_demand_loop"):
|
|
parts.append(json.dumps(proposal.token_demand_loop.loop, default=str))
|
|
crypto = proposal.metadata.get("crypto", {}) if isinstance(proposal.metadata, dict) else {}
|
|
actual = {key: crypto.get(key) for key in ["product_thesis", "protocol_thesis", "token_thesis", "token_utility", "token_demand_loop", "value_capture", "network_effect", "bootstrap_plan", "security_model", "token_roles", "mechanism_fingerprint"] if key in crypto}
|
|
parts.append(json.dumps(actual, default=str))
|
|
return " ".join(str(part) for part in parts if part).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 {}
|
|
return self._model_json(self.final_ic_model_hint, prompt, purpose=ModelCapability.REASONING)
|
|
|
|
def _model_json(self, model_hint: str, prompt: str, *, purpose: str) -> dict[str, Any]:
|
|
if self.router is None or model_hint not in self.router.providers:
|
|
return {}
|
|
try:
|
|
response = self.router.complete(ModelRequestContract(purpose=purpose, model_hint=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.3.1", "", 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)
|