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

673 lines
56 KiB
Python
Raw Normal View History

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