1396 lines
139 KiB
Python
1396 lines
139 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from itertools import combinations
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from collections import Counter, defaultdict
|
|
from django.core.exceptions import ObjectDoesNotExist
|
|
from django.db import close_old_connections, connection
|
|
from django.utils import timezone
|
|
|
|
from control_plane.events.bus import EventBus
|
|
from control_plane.ventures.models import AutonomousCandidateStatus, AutonomousGateResult, AutonomousOperabilityAssessment, CapabilityPriority, CapabilityStatus, CohortIdeationMandate, CompanyBoardReview, CompanyCapabilityRequirement, CompanyMandate, CompanyProposal, CompanyProposalStatus, EvidenceTier, FounderDependencyLevel, ICDecision, ICDecisionType, ICDiligence, ICQuestion, ICResponse, NoveltyGateDecision, OpportunityTerritory, OverlapClassification, PortfolioCapabilityGap, PortfolioICReview, PortfolioSaturationAnalysis, PortfolioThesis, PortfolioThesisCluster, PortfolioThesisStatus, ThesisMatchClassification, VentureArtifact, VentureCapabilityDemand, VentureCohort, VentureCohortMember, VentureCollision, VentureGenerationRejection, VentureThesis, VentureThesisFingerprint, VentureTrack
|
|
from graph.models import GraphRun, GraphRunStatus
|
|
from model_router.providers import extract_json_object
|
|
from model_router.policy import model_for_role
|
|
from model_router.router import ModelCapability, ModelRequestContract, ModelRouter
|
|
from research.searxng import SearxngSearchClient, WebPageFetcher
|
|
|
|
|
|
PITCH_SECTIONS = ["Company name", "One-line thesis", "Problem", "ICP", "Why now", "Product / service", "Business model", "Pricing", "Route to first customer", "Validation plan", "$50 capital allocation proposal", "Time to first dollar", "Path to $500 net cash", "Competition", "Differentiation", "Build requirements", "Distribution requirements", "Risks", "What would falsify the thesis", "Confidence"]
|
|
SCORE_DIMENSIONS = ["Demand Evidence", "Time-to-First-Dollar Attractiveness", "Capital Efficiency", "Validation Affordability", "Gross Margin Potential", "Distribution Feasibility", "Build Simplicity", "Defensibility", "Market Opportunity", "Competitive Position", "Risk Manageability", "AI Leverage", "Platformization Potential", "Autonomous Operability", "Probability of Reaching $500"]
|
|
SCORE_DEFINITIONS = {dimension: "100 = highly attractive; 0 = highly unattractive" for dimension in SCORE_DIMENSIONS}
|
|
EVIDENCE_CEILINGS = {EvidenceTier.TIER_0_THESIS: 35, EvidenceTier.TIER_1_PUBLIC_EVIDENCE: 55, EvidenceTier.TIER_2_CUSTOMER_SIGNAL: 70, EvidenceTier.TIER_3_WILLINGNESS_TO_PAY: 85, EvidenceTier.TIER_4_PAID_CUSTOMER: 95, EvidenceTier.TIER_5_REPEATABLE_TRACTION: 100}
|
|
AI_NATIVE_POLICY = {"preference": "Prefer AI-native opportunities where Artifex can build or deliver most value with existing agents, local inference, and owned workflow capabilities.", "preferred_classes": ["AI wrapper platforms", "AI-enabled services", "AI infrastructure / developer tools", "AI research / intelligence products", "AI automation for SMB / enterprise workflows"], "soft_portfolio_targets": {"ai_wrapper_platforms": "40-50%", "ai_enabled_services": "20-30%", "developer_infrastructure_intelligence": "10-20%", "non_ai_economic_outliers": "10-20%"}, "penalize_concentration": ["Shopify/ecommerce", "generic audits", "emergency fix services", "one-off consulting", "identical outbound-led service models"], "do_not": "Do not let AI novelty outweigh customer demand."}
|
|
DEFAULT_HARD_EXCLUSIONS = ["AI RFP response drafting / proposal automation", "SaaS churn prediction / retention playbook generation"]
|
|
DEFAULT_SOFT_EXCLUSIONS = ["generic contract scanners", "Shopify audit products", "generic AI content generators", "generic productized consulting", "generic one-off audits", "generic chatbot wrappers"]
|
|
DEFAULT_TERRITORY_ALLOCATION = [OpportunityTerritory.VERTICAL_AI_WRAPPERS, OpportunityTerritory.VERTICAL_AI_WRAPPERS, OpportunityTerritory.ENTERPRISE_WORKFLOW_AUTOMATION, OpportunityTerritory.SMB_AUTOMATION, OpportunityTerritory.DEVELOPER_AI_INFRASTRUCTURE, OpportunityTerritory.INTELLIGENCE_MONITORING, OpportunityTerritory.DATA_DOCUMENT_AUTOMATION, OpportunityTerritory.AI_ENABLED_SERVICE_TO_PLATFORM, OpportunityTerritory.OPEN_CATEGORY, OpportunityTerritory.OPEN_CATEGORY]
|
|
DEFAULT_DUPLICATE_POLICY = {"max_near_duplicate_per_thesis_per_cohort": 1, "max_competitive_per_thesis_per_cohort": 2, "max_attempts_per_slot": 3, "cohort_attempt_budget_multiplier": 4}
|
|
RESEARCH_CATEGORIES = ["competitors", "pricing", "customer_pain", "market_alternatives", "regulatory_platform_risks"]
|
|
AUTONOMOUS_ARCHETYPES = ["SELF_SERVICE_AI_TOOL", "AUTOMATED_MONITORING_PRODUCT", "DIGITAL_ANALYSIS_SERVICE", "AUTOMATED_TRANSFORMATION_SERVICE", "MICRO_SAAS", "DATA_INTELLIGENCE_SUBSCRIPTION", "DEVELOPER_TOOL", "AUTONOMOUS_DIGITAL_PRODUCT", "MARKETPLACE_DELIVERED_SERVICE", "AGENT_AS_A_SERVICE"]
|
|
|
|
|
|
class VentureDiscoveryService:
|
|
def __init__(self, router: ModelRouter | None = None, bus: EventBus | None = None, web_research_available: bool = False, research_model_hint: str | None = None, ideation_model_hint: str | None = None, search_client: Any | None = None, page_fetcher: Any | None = None) -> None:
|
|
self.router = router
|
|
self.bus = bus or EventBus()
|
|
self.web_research_available = web_research_available
|
|
self.research_model_hint = research_model_hint or model_for_role("venture_research")
|
|
self.ideation_model_hint = ideation_model_hint or model_for_role("venture_ideation")
|
|
self.search_client = search_client
|
|
self.page_fetcher = page_fetcher or WebPageFetcher()
|
|
self._last_bounded_map_peak = 0
|
|
|
|
def create_v0_mandate(self) -> CompanyMandate:
|
|
mandate = CompanyMandate.objects.create(
|
|
objective="Design a business that Artifex itself could plausibly operate and turn at most $50 of external validation capital into at least $500 net new cash within 30 days, with no more than 30 minutes/week of routine human intervention.",
|
|
constraints={"external_validation_capital_max": 50, "target_net_new_cash": 500, "target_window_days": 30, "max_human_routine_minutes_per_week": 30, "default_venture_track": VentureTrack.AUTONOMOUS, "routine_human_involvement_not_allowed": ["founder-led sales calls", "manual prospecting", "routine support", "manual fulfillment", "bespoke consulting", "routine QA", "manual payment chasing", "routine onboarding"], "allowed_human_involvement": ["approve account/legal setup", "approve initial spend", "approve material legal commitments", "exceptional safety/security escalation", "irreversible capital decisions"], "no_equity_raise": True, "no_debt": True, "no_illegal_or_deceptive_activity": True, "no_spam": True, "no_fake_traction": True, "no_fabricated_customer_evidence": True, "no_real_spend_in_v0": True, "no_real_customer_outreach_in_v0": True, "existing_artifex_compute_sunk_available": True, "validation_deployment": "SUBDOMAIN_DEPLOYMENT"},
|
|
optimization_targets=["autonomous fulfillment", "autonomous customer acquisition", "self-service onboarding", "standardized digital delivery", "AI leverage", "platformization", "recurring revenue", "simple payment flow", "low support burden", "low regulatory burden", "low integration burden", "rapid validation"],
|
|
metadata={"milestone": "VENTURE_DISCOVERY_V0_AUTONOMOUS", "venture_track": VentureTrack.AUTONOMOUS, "spend_authorized": False, "customer_outreach_authorized": False, "ai_native_policy": AI_NATIVE_POLICY, "autonomous_archetypes": AUTONOMOUS_ARCHETYPES},
|
|
)
|
|
self._artifact(None, mandate, "VENTURE_MANDATE", "Venture Discovery V0 Mandate", {"objective": mandate.objective, "constraints": mandate.constraints, "optimization_targets": mandate.optimization_targets}, self._readable_mandate(mandate), "venture_discovery")
|
|
return mandate
|
|
|
|
def generate_single_company(self, mandate: CompanyMandate, *, graph_run=None, ideation_index: int | None = None, ideation_context: dict[str, Any] | None = None, payload: dict[str, Any] | None = None, source: str | None = None) -> CompanyProposal:
|
|
if payload is None or source is None:
|
|
payload, source = self._company_payload(mandate, ideation_index=ideation_index, ideation_context=ideation_context)
|
|
return self._create_company_from_payload(mandate, payload, source, graph_run=graph_run)
|
|
|
|
def _create_company_from_payload(self, mandate: CompanyMandate, payload: dict[str, Any], source: str, *, graph_run=None) -> CompanyProposal:
|
|
pitch = self._pitch(payload, fallback=source == "deterministic_fallback")
|
|
confidence = self._confidence(payload.get("confidence", 0.55))
|
|
evidence = self._as_list(payload.get("market_evidence", []))
|
|
if not self.web_research_available:
|
|
evidence.append({"type": "capability_gap", "source": "internal", "summary": "Public web market research is not configured; demand/competitor evidence is unverified.", "fallback_evidence": True})
|
|
confidence = min(confidence, 0.58)
|
|
thesis = VentureThesis.objects.create(mandate=mandate, title=str(payload["title"]), thesis=str(payload["one_line_thesis"]), similarity_fingerprint=self._fingerprint(payload), metadata={"source": source, "exactly_one_company_generated": True}, evidence_tier=EvidenceTier.TIER_0_THESIS)
|
|
proposal = CompanyProposal.objects.create(
|
|
mandate=mandate,
|
|
thesis=thesis,
|
|
title=str(payload["title"]),
|
|
description=str(payload["description"]),
|
|
problem=str(payload["problem"]),
|
|
target_customer=str(payload["target_customer"]),
|
|
proposed_solution=str(payload["proposed_solution"]),
|
|
business_model=str(payload["business_model"]),
|
|
pricing_hypothesis=str(payload["pricing_hypothesis"]),
|
|
acquisition_strategy=str(payload["acquisition_strategy"]),
|
|
validation_plan=str(payload["validation_plan"]),
|
|
capital_requested=self._money(payload.get("capital_requested", "50")),
|
|
time_to_first_dollar_estimate=str(payload["time_to_first_dollar_estimate"]),
|
|
expected_margin=str(payload["expected_margin"]),
|
|
build_complexity=str(payload["build_complexity"]),
|
|
market_evidence=evidence,
|
|
differentiation=str(payload["differentiation"]),
|
|
major_risks=self._as_list(payload["major_risks"]),
|
|
confidence=confidence,
|
|
status=CompanyProposalStatus.SUBMITTED,
|
|
pitch=pitch,
|
|
metadata={"generation_source": source, "fallback_evidence": source == "deterministic_fallback", "web_research_available": self.web_research_available, "real_spend": 0, "real_customer_outreach": False, "validation_offer": payload.get("validation_offer", {}), "end_state_business_model": payload.get("end_state_business_model", {}), "fulfillment_contract": payload.get("fulfillment_contract", {}), "minutes_per_week_human": payload.get("minutes_per_week_human", None), "human_actions_required": self._as_list(payload.get("human_actions_required", []))},
|
|
evidence_tier=EvidenceTier.TIER_0_THESIS,
|
|
)
|
|
self._artifact(proposal, mandate, "STANDARDIZED_COMPANY_PITCH", "Standardized Company Pitch", pitch, self.readable_pitch(pitch), f"Company Brain/{source}" if source != "deterministic_fallback" else "deterministic_fallback", graph_run=graph_run)
|
|
self.bus.publish("VENTURE_COMPANY_PROPOSED", payload={"proposal_id": str(proposal.id), "source": source})
|
|
return proposal
|
|
|
|
def conduct_market_research(self, proposal: CompanyProposal, *, graph_run=None, depth: str = "light") -> dict[str, Any]:
|
|
required = RESEARCH_CATEGORIES
|
|
research = {"coverage": {category: False for category in required}, "sources": [], "findings": {}, "unverified_categories": required, "research_available": False}
|
|
search_sources = self._searxng_sources(proposal, required, depth=depth) if self.web_research_available else []
|
|
page_corpus = self._page_corpus(search_sources) if search_sources else []
|
|
diagnostics = {"search_result_count": len(search_sources), "page_fetch_count": len(page_corpus), "model_provider": self.research_model_hint if self.router is not None else "none", "search_provider": "searxng" if search_sources else "none"}
|
|
search_diagnostics = getattr(self, "_last_search_diagnostics", {})
|
|
if search_diagnostics:
|
|
diagnostics["search_diagnostics"] = search_diagnostics
|
|
if self.web_research_available and self.router is not None:
|
|
try:
|
|
response = self.router.complete(ModelRequestContract(purpose=ModelCapability.REASONING, model_hint=self.research_model_hint, prompt="Bounded public web market research for exactly one startup pitch. Return JSON with keys: sources (list of {url,title,category,summary}), findings (object keyed by competitors, pricing, customer_pain, market_alternatives, regulatory_platform_risks), and coverage (object with each required key true/false). Use only the provided SearXNG search context and fetched page excerpts for source URLs; do not fabricate URLs. If a category has no strong relevant source, mark coverage false. Research depth: " + depth + ". Search context: " + json.dumps(search_sources, default=str) + " Page excerpts: " + json.dumps(page_corpus, default=str) + " Pitch: " + json.dumps(proposal.pitch, default=str)))
|
|
parsed = extract_json_object(response.content)
|
|
if isinstance(parsed, dict):
|
|
research = self._normalize_research(parsed, required)
|
|
research["research_available"] = True
|
|
research["provider"] = self.research_model_hint
|
|
except Exception as exc:
|
|
research["failure"] = str(exc)
|
|
if search_sources:
|
|
research = self._merge_search_sources(research, search_sources, required)
|
|
if page_corpus:
|
|
research = {**research, "page_corpus": page_corpus, "page_fetch_count": len(page_corpus)}
|
|
if not research.get("sources"):
|
|
research = {**self._missing_research(required, research.get("failure", "public web research unavailable or returned no source-linked evidence")), **diagnostics}
|
|
quality = self.filter_research_sources(proposal, research.get("sources", []), required)
|
|
research["sources"] = quality["accepted_sources"]
|
|
research["source_rejections"] = quality["rejected_sources"]
|
|
research["source_rejection_count"] = len(quality["rejected_sources"])
|
|
source_categories = {str(source.get("category", "")) for source in research.get("sources", []) if source.get("url") and source.get("quality") == "strong"}
|
|
coverage = dict(research.get("coverage", {}))
|
|
for category in required:
|
|
coverage[category] = bool(coverage.get(category)) and category in source_categories
|
|
research["coverage"] = coverage
|
|
research["unverified_categories"] = [category for category in required if not coverage.get(category)]
|
|
coverage_ratio = (len(required) - len(research["unverified_categories"])) / len(required)
|
|
proposal.confidence = round(min(float(proposal.confidence), 0.45 + 0.35 * coverage_ratio), 2)
|
|
existing_research = proposal.metadata.get("research", {}) if isinstance(proposal.metadata, dict) else {}
|
|
proposal.market_evidence = [*self._as_list(proposal.market_evidence), *research.get("sources", []), {"type": "research_coverage", "source": "venture_research", "summary": f"{depth.title()} source-linked research coverage: {round(coverage_ratio * 100)}%", "coverage": coverage, "unverified_categories": research["unverified_categories"], "depth": depth}]
|
|
explicit_failure = "" if research.get("sources") else research.get("failure", "public web research unavailable or returned no source-linked evidence")
|
|
proposal.metadata = {**proposal.metadata, "research": {**existing_research, "coverage_ratio": coverage_ratio, "coverage": coverage, "unverified_categories": research["unverified_categories"], "source_count": len(research.get("sources", [])), "source_rejection_count": len(quality["rejected_sources"]), "search_result_count": research.get("search_result_count", len(search_sources)), "page_fetch_count": research.get("page_fetch_count", len(page_corpus)), "provider": research.get("provider", diagnostics.get("model_provider", "none")), "search_provider": research.get("search_provider", diagnostics["search_provider"]), "search_diagnostics": research.get("search_diagnostics", diagnostics.get("search_diagnostics", {})), "model_research_failure": research.get("failure", "") if research.get("sources") else "", "depth": depth, "explicit_research_failure": explicit_failure, "before_deep_coverage_ratio": existing_research.get("coverage_ratio") if depth == "deep" else existing_research.get("before_deep_coverage_ratio")}}
|
|
proposal.evidence_tier = EvidenceTier.TIER_1_PUBLIC_EVIDENCE if coverage_ratio > 0 else EvidenceTier.TIER_0_THESIS
|
|
if proposal.thesis:
|
|
proposal.thesis.evidence_tier = proposal.evidence_tier
|
|
proposal.thesis.save(update_fields=["evidence_tier", "updated_at"])
|
|
proposal.pitch = {**proposal.pitch, "Research evidence": research}
|
|
proposal.save(update_fields=["confidence", "market_evidence", "metadata", "pitch", "evidence_tier", "updated_at"])
|
|
self._artifact(proposal, proposal.mandate, "MARKET_RESEARCH", f"Bounded Public Market Research ({depth})", research, self._readable_research(research), f"{research.get('provider', 'search')}/web research" if research.get("sources") else "research_gap", graph_run=graph_run)
|
|
return research
|
|
|
|
def board_review(self, proposal: CompanyProposal, *, graph_run=None) -> CompanyBoardReview:
|
|
observations = {
|
|
"CEO": ["Mandate fit is strongest if validation sells a narrow paid audit before product build.", "Keep the first dollar path service-led, not SaaS-led."],
|
|
"CTO": ["Build can use existing Artifex analysis, reporting, and frontend capabilities.", "Avoid integrations until willingness-to-pay evidence exists."],
|
|
"CFO": ["$50 cap is adequate only for lightweight landing page/listing tests, not paid acquisition learning.", "High margin is plausible because delivery is mostly labor/compute already available."],
|
|
"CRO": ["Founder/operator communities are reachable manually, but V0 cannot contact customers.", "Pricing must start as a paid diagnostic to avoid long SaaS evaluation cycles."],
|
|
"Independent Director": ["Demand evidence is weak without public research or customer conversations.", "The company must prove urgency before building automation."],
|
|
}
|
|
weaknesses = ["No customer outreach or paid test has occurred.", "Public web research is unavailable, so market evidence remains partial.", "First customers may require trust and examples before paying."]
|
|
revisions = ["Frame offer as a productized validation audit with optional Artifex-assisted build plan.", "Make falsification criteria explicit before spend."]
|
|
pitch = dict(proposal.pitch)
|
|
pitch["What would falsify the thesis"] = "Fewer than 5 credible target-customer responses or zero willingness-to-pay signals after a compliant validation test."
|
|
proposal.pitch = pitch
|
|
proposal.metadata = {**proposal.metadata, "board_revised_pitch": True}
|
|
proposal.save(update_fields=["pitch", "metadata", "updated_at"])
|
|
identity = self.validate_identity_content(proposal, pitch)
|
|
review = CompanyBoardReview.objects.create(proposal=proposal, observations=observations, strengths=["Service-led revenue path can precede product build.", "Uses current Artifex planning, engineering, and review capabilities.", "Small validation budget aligns with a narrow paid offer."], weaknesses=weaknesses, key_assumptions=["Target customers feel enough urgency to pay for validation clarity.", "Manual outbound or community posting can generate credible responses without spam.", "Artifex can produce a differentiated audit faster than generic consultants."], required_revisions=revisions, recommendation="PROCEED_TO_IC_WITH_REVISIONS", revised_pitch=pitch, metadata={"roles": list(observations), "company_does_not_grade_itself": True, "identity_validation": identity})
|
|
self._artifact(proposal, proposal.mandate, "COMPANY_BOARD_REVIEW", "Company Board Review", {"observations": observations, "strengths": review.strengths, "weaknesses": weaknesses, "required_revisions": revisions, "recommendation": review.recommendation, "revised_pitch": pitch, "identity_validation": identity}, self._readable_board(review), "Company Board", graph_run=graph_run)
|
|
return review
|
|
|
|
def start_ic_diligence(self, proposal: CompanyProposal, *, graph_run=None) -> ICDiligence:
|
|
proposal.status = CompanyProposalStatus.UNDER_DILIGENCE
|
|
proposal.save(update_fields=["status", "updated_at"])
|
|
diligence = ICDiligence.objects.create(proposal=proposal, status="FIRST_PASS", rounds=["Initial Pitch", "Diligence Round 1", "Final Challenge", "Decision"], metadata={"bounded_rounds": True, "self_grading": False})
|
|
self._artifact(proposal, proposal.mandate, "IC_FIRST_PASS", "IC First-Pass Review", {"status": diligence.status, "initial_concerns": ["Demand evidence is unverified.", "Distribution assumptions need evidence.", "Need validation gate before any spend."]}, "IC first pass: proceed to evidence-seeking questions; do not score final decision yet.", "Independent IC", graph_run=graph_run)
|
|
return diligence
|
|
|
|
def generate_ic_questions(self, diligence: ICDiligence, *, graph_run=None) -> list[ICQuestion]:
|
|
pitch = diligence.proposal.pitch
|
|
raw_questions = self._questions_from_pitch(pitch)
|
|
questions = [ICQuestion.objects.create(diligence=diligence, question=item["question"], category=item["category"], evidence_required=True) for item in raw_questions[:8]]
|
|
self._artifact(diligence.proposal, diligence.proposal.mandate, "IC_QUESTIONS", "IC Diligence Questions", {"questions": [q.question for q in questions]}, "\n".join(f"- {q.question}" for q in questions), "Independent IC", graph_run=graph_run)
|
|
return questions
|
|
|
|
def answer_questions(self, diligence: ICDiligence, *, graph_run=None) -> list[ICResponse]:
|
|
responses = []
|
|
for question in diligence.questions.all():
|
|
answer = self._answer_question(question)
|
|
responses.append(ICResponse.objects.create(question=question, answer=answer["answer"], evidence=answer["evidence"], uncertainty=answer["uncertainty"], pitch_changes=answer.get("pitch_changes", {}), metadata={"no_customer_outreach": True, "no_spend": True}))
|
|
diligence.status = "COMPANY_RESPONSE"
|
|
diligence.save(update_fields=["status", "updated_at"])
|
|
identity = self.validate_identity_content(diligence.proposal, {"responses": [r.answer for r in responses]})
|
|
self._artifact(diligence.proposal, diligence.proposal.mandate, "IC_RESPONSES", "Company Responses to IC", {"responses": [{"question": r.question.question, "answer": r.answer, "evidence": r.evidence, "uncertainty": r.uncertainty, "pitch_changes": r.pitch_changes} for r in responses], "identity_validation": identity}, self._readable_responses(responses), "Company reasoning roles", graph_run=graph_run)
|
|
return responses
|
|
|
|
def red_team(self, diligence: ICDiligence, *, graph_run=None) -> dict[str, object]:
|
|
challenge = {"concerns": ["The offer may be perceived as generic consulting unless anchored to a painful, immediate decision.", "Without public research or customer contact, demand remains a hypothesis.", "Manual distribution could fail if the target audience distrusts AI-generated audits."], "required_final_response": ["Narrow ICP further.", "Specify willingness-to-pay proof.", "Define hard kill criteria."], "recommendation": "continue_to_final_response"}
|
|
diligence.red_team_challenge = challenge
|
|
diligence.status = "FINAL_CHALLENGE"
|
|
diligence.save(update_fields=["red_team_challenge", "status", "updated_at"])
|
|
challenge["identity_validation"] = self.validate_identity_content(diligence.proposal, challenge)
|
|
self._artifact(diligence.proposal, diligence.proposal.mandate, "IC_RED_TEAM", "IC Red-Team Challenge", challenge, "Red-team concerns:\n" + "\n".join(f"- {c}" for c in challenge["concerns"]), "Independent IC Red Team", graph_run=graph_run)
|
|
return challenge
|
|
|
|
def final_company_response(self, diligence: ICDiligence, *, graph_run=None) -> dict[str, object]:
|
|
response = {"narrowed_icp": diligence.proposal.target_customer, "revised_validation_gate": "Before any spend, collect 5 credible target-customer responses or 1 explicit willingness-to-pay signal through compliant non-spam channels.", "kill_criteria": ["No credible responses after 10 targeted, compliant conversations/posts once outreach is approved.", "No willingness-to-pay signal at $49-$99.", "Customers only want free advice, not a paid report."], "pitch_changes": {"ICP": "Preserve canonical proposal ICP.", "Validation plan": "Gate spend behind credible response/willingness-to-pay evidence."}}
|
|
response["identity_validation"] = self.validate_identity_content(diligence.proposal, response)
|
|
diligence.final_response = response
|
|
diligence.status = "FINAL_RESPONSE"
|
|
diligence.save(update_fields=["final_response", "status", "updated_at"])
|
|
self._artifact(diligence.proposal, diligence.proposal.mandate, "IC_FINAL_RESPONSE", "Final Company Response", response, json.dumps(response, indent=2), "Company reasoning roles", graph_run=graph_run)
|
|
return response
|
|
|
|
def score_and_decide(self, diligence: ICDiligence, *, graph_run=None) -> ICDecision:
|
|
try:
|
|
assessment = diligence.proposal.autonomous_assessment
|
|
except ObjectDoesNotExist:
|
|
assessment = None
|
|
scores = self._evidence_scores(diligence.proposal)
|
|
if assessment is not None:
|
|
scores["Autonomous Operability"] = int(assessment.autonomous_operability_score)
|
|
composite = round(sum(scores.values()) / len(scores), 1)
|
|
calibration = self.calibrate_probability(diligence.proposal, raw_probability=float(scores["Probability of Reaching $500"]))
|
|
scores["Probability of Reaching $500"] = int(calibration["evidence_adjusted_probability"])
|
|
composite = round(sum(scores.values()) / len(scores), 1)
|
|
decision_type = ICDecisionType.REVISE_AND_RESUBMIT if scores["Demand Evidence"] < 35 or scores["Risk Manageability"] < 35 else ICDecisionType.CONDITIONAL_FUND if composite >= 55 else ICDecisionType.REVISE_AND_RESUBMIT
|
|
tranche = Decimal("20.00") if decision_type == ICDecisionType.CONDITIONAL_FUND and composite >= 70 else Decimal("10.00") if decision_type == ICDecisionType.CONDITIONAL_FUND else None
|
|
decision, _ = ICDecision.objects.update_or_create(diligence=diligence, defaults={"decision": decision_type, "component_scores": scores, "composite_score": composite, "probability_500_within_30_days": scores["Probability of Reaching $500"], "initial_tranche": tranche, "validation_condition": "Obtain 5 credible target-customer responses or 1 explicit willingness-to-pay signal before any build or further spend.", "evidence_required": ["response transcripts or public thread URLs", "proof of willingness-to-pay signal", "no-spam/no-fabrication compliance note"], "recommended_allocation": {"initial_tranche": float(tranche or 0), "remaining_reserved": 50 - float(tranche or 0), "no_spend_in_v0": True}, "kill_criteria": diligence.final_response.get("kill_criteria", []), "next_decision_point": "After validation evidence is collected and before any real spend or customer delivery.", "metadata": {"decision_vocabulary": [item.value for item in ICDecisionType], "no_actual_funding": True, "score_basis": "source_linked_research_and_pitch_attributes", "probability_calibration": calibration}, "evidence_tier": diligence.proposal.evidence_tier, "raw_probability_500_within_30_days": calibration["raw_probability"], "evidence_ceiling": calibration["evidence_ceiling"], "probability_explanation": calibration["explanation"], "score_definitions": SCORE_DEFINITIONS})
|
|
diligence.status = "DECISION"
|
|
diligence.save(update_fields=["status", "updated_at"])
|
|
diligence.proposal.status = CompanyProposalStatus.FUNDED_RECOMMENDED if decision.decision == ICDecisionType.CONDITIONAL_FUND else CompanyProposalStatus.REVISE
|
|
diligence.proposal.save(update_fields=["status", "updated_at"])
|
|
self._artifact(diligence.proposal, diligence.proposal.mandate, "IC_FINAL_SCORE", "IC Final Scoring and Decision", {"scores": scores, "score_definitions": SCORE_DEFINITIONS, "decision": decision.decision, "composite_score": composite, "conditional_funding": {"initial_tranche": str(decision.initial_tranche), "condition": decision.validation_condition}, "score_basis": decision.metadata["score_basis"], "probability_calibration": calibration}, self._readable_score(decision), "Independent IC", graph_run=graph_run)
|
|
return decision
|
|
|
|
def capability_analysis(self, proposal: CompanyProposal, *, graph_run=None) -> PortfolioCapabilityGap:
|
|
requirements = self._capability_requirements(proposal)
|
|
for item in requirements:
|
|
CompanyCapabilityRequirement.objects.create(proposal=proposal, **item)
|
|
available = [r["category"] for r in requirements if r["status"] == CapabilityStatus.AVAILABLE]
|
|
partial = [r["category"] for r in requirements if r["status"] == CapabilityStatus.PARTIAL]
|
|
missing = [r["category"] for r in requirements if r["status"] == CapabilityStatus.MISSING]
|
|
ranked = [{"category": r["category"], "priority": r["priority"], "rationale": r["rationale"]} for r in requirements if r["status"] == CapabilityStatus.MISSING]
|
|
priority_order = {CapabilityPriority.BEFORE_VALIDATION: 0, CapabilityPriority.BEFORE_FIRST_CUSTOMER: 1, CapabilityPriority.BEFORE_SCALING: 2}
|
|
ranked.sort(key=lambda item: priority_order[item["priority"]])
|
|
report = self._readable_capability_gap(available, partial, missing, ranked)
|
|
web_requirement = next((item for item in requirements if item["category"] == "WEB_MARKET_RESEARCH"), None)
|
|
gap = PortfolioCapabilityGap.objects.create(proposal=proposal, available=available, partial=partial, missing=missing, ranked_missing=ranked, report=report, metadata={"web_market_research_status": web_requirement["status"] if web_requirement else "MISSING"})
|
|
self._artifact(proposal, proposal.mandate, "CAPABILITY_GAP_REPORT", "Capability Gap Report", {"available": available, "partial": partial, "missing": missing, "ranked_missing": ranked}, report, "Venture Discovery Capability Analysis", graph_run=graph_run)
|
|
return gap
|
|
|
|
def produce_investment_memo(self, diligence: ICDiligence, gap: PortfolioCapabilityGap, *, graph_run=None) -> VentureArtifact:
|
|
decision = diligence.decision
|
|
proposal = diligence.proposal
|
|
memo = {"Company": proposal.title, "Thesis": proposal.pitch["One-line thesis"], "Mandate": proposal.mandate.objective, "Requested capital": str(proposal.capital_requested), "Recommended allocation": decision.recommended_allocation, "IC decision": decision.decision, "Key metrics": {"P($500 within 30 days)": decision.probability_500_within_30_days, "raw P($500 within 30 days)": decision.raw_probability_500_within_30_days, "evidence ceiling": decision.evidence_ceiling, "evidence tier": decision.evidence_tier, "AI leverage": decision.component_scores.get("AI Leverage"), "platformization potential": decision.component_scores.get("Platformization Potential"), "estimated time to first dollar": proposal.time_to_first_dollar_estimate, "expected gross margin": proposal.expected_margin, "validation cost": "$10-$20 initial tranche; $50 maximum after approval", "build effort": proposal.build_complexity, "distribution feasibility": decision.component_scores["Distribution Feasibility"]}, "Research evidence": proposal.pitch.get("Research evidence", {}), "Why it may work": ["Revenue path starts with a paid diagnostic, not a full SaaS build.", "Artifex has planning, engineering, frontend, review, graph, and agent-control capabilities already.", "Validation budget can be gated behind evidence."], "Why it may fail": diligence.red_team_challenge.get("concerns", []), "Diligence questions": [q.question for q in diligence.questions.all()], "Company responses": [r.answer for r in ICResponse.objects.filter(question__diligence=diligence)], "Red-team concerns": diligence.red_team_challenge.get("concerns", []), "IC scoring": decision.component_scores, "Score definitions": decision.score_definitions, "Capital recommendation": decision.recommended_allocation, "Validation gates": [decision.validation_condition], "Kill criteria": decision.kill_criteria, "Next decision point": decision.next_decision_point, "Capability gap": {"available": gap.available, "partial": gap.partial, "missing": gap.missing}}
|
|
identity = self.validate_identity_content(proposal, memo)
|
|
memo["Identity validation"] = identity
|
|
return self._artifact(proposal, proposal.mandate, "FINAL_INVESTMENT_MEMO", "Final IC Investment Memo", memo, self._readable_memo(memo), "Independent IC", graph_run=graph_run)
|
|
|
|
def request_human_approval(self, decision: ICDecision, action: str) -> ICDecision:
|
|
if action not in {"approve_for_validation", "reject", "request_more_diligence"}:
|
|
raise ValueError("Unsupported venture approval action")
|
|
decision.metadata = {**decision.metadata, "human_approval_action": action, "real_spend_still_blocked": True}
|
|
decision.save(update_fields=["metadata", "updated_at"])
|
|
return decision
|
|
|
|
def prepare_cohort(self, *, size: int = 10, graph_run=None, concurrency: int = 1) -> VentureCohort:
|
|
self.seed_dogfood_thesis_registry()
|
|
mandate = self.create_v0_mandate()
|
|
cohort_id = f"VDV04-{timezone.now().strftime('%Y%m%d%H%M%S')}-{hashlib.sha1(str(mandate.id).encode()).hexdigest()[:8]}"
|
|
return VentureCohort.objects.create(cohort_id=cohort_id, mandate=mandate, cohort_size=size, graph_run=graph_run, concurrency=concurrency, status="PREPARING", graph_versions={"cohort": "venture_discovery_cohort v3", "company": "venture_discovery v1"}, research_policy={"stage_a": "lightweight_all", "finalist_deeper_research": "top_5", "target_finalist_coverage": 0.8, "no_spend": True, "no_customer_outreach": True}, scoring_policy={"dimensions": SCORE_DEFINITIONS, "duplicate_policy": DEFAULT_DUPLICATE_POLICY}, evidence_calibration_policy={tier: ceiling for tier, ceiling in EVIDENCE_CEILINGS.items()}, metadata={"real_spend": 0, "real_customer_outreach": False, "milestone": "VENTURE_DISCOVERY_V0.4"})
|
|
|
|
def portfolio_thesis_review(self, cohort: VentureCohort) -> dict[str, Any]:
|
|
self.seed_dogfood_thesis_registry()
|
|
registry = self.registry_snapshot()
|
|
review = {"saturated_thesis_areas": [row for row in registry if row["status"] == PortfolioThesisStatus.SATURATED], "active_candidates": [row for row in registry if row["status"] == PortfolioThesisStatus.ACTIVE_CANDIDATE], "registry_size": len(registry), "policy": DEFAULT_DUPLICATE_POLICY}
|
|
cohort.metadata = {**cohort.metadata, "portfolio_thesis_review": review}
|
|
cohort.save(update_fields=["metadata", "updated_at"])
|
|
return review
|
|
|
|
def create_ideation_mandate(self, cohort: VentureCohort) -> CohortIdeationMandate:
|
|
territories = self.allocate_territories(cohort.cohort_size)
|
|
review = cohort.metadata.get("portfolio_thesis_review", {}) if isinstance(cohort.metadata, dict) else {}
|
|
objective = "Intentionally search new venture territory while preserving independent ideation and avoiding known saturated thesis areas."
|
|
mandate, _ = CohortIdeationMandate.objects.update_or_create(
|
|
cohort=cohort,
|
|
defaults={
|
|
"objective": objective,
|
|
"desired_opportunity_classes": AI_NATIVE_POLICY["preferred_classes"],
|
|
"hard_exclusions": DEFAULT_HARD_EXCLUSIONS,
|
|
"soft_exclusions": DEFAULT_SOFT_EXCLUSIONS,
|
|
"opportunity_territories": territories,
|
|
"portfolio_gaps": ["non-RFP enterprise workflows", "non-churn SaaS intelligence", "developer infrastructure", "document/data automation"],
|
|
"diversity_preferences": {"avoid_repeating_saturated_theses": True, "max_near_duplicate_per_thesis_per_cohort": 1, "max_competitive_per_thesis_per_cohort": 2, "ai_native_preferred_not_required": True},
|
|
"registry_snapshot": {"version": timezone.now().isoformat(), "theses": self.registry_snapshot()},
|
|
"metadata": {"portfolio_thesis_review": review, "cohort_ideation_brief": self.ideation_brief_text(review, territories)},
|
|
},
|
|
)
|
|
self._artifact(None, cohort.mandate, "COHORT_IDEATION_MANDATE", "Cohort Ideation Mandate", self.ideation_mandate_payload(mandate), mandate.metadata["cohort_ideation_brief"], "Portfolio IC", graph_run=cohort.graph_run)
|
|
return mandate
|
|
|
|
def allocate_search_territories(self, cohort: VentureCohort) -> list[str]:
|
|
mandate = getattr(cohort, "ideation_mandate", None) or self.create_ideation_mandate(cohort)
|
|
territories = self.allocate_territories(cohort.cohort_size)
|
|
mandate.opportunity_territories = territories
|
|
mandate.save(update_fields=["opportunity_territories", "updated_at"])
|
|
return territories
|
|
|
|
def generate_independent_proposals(self, cohort: VentureCohort) -> list[CompanyProposal]:
|
|
started = time.monotonic()
|
|
accepted_payloads: list[dict[str, Any]] = []
|
|
proposals = []
|
|
mandate = getattr(cohort, "ideation_mandate", None) or self.create_ideation_mandate(cohort)
|
|
policy = cohort.scoring_policy.get("duplicate_policy", DEFAULT_DUPLICATE_POLICY)
|
|
total_budget = max(cohort.cohort_size, int(cohort.cohort_size * int(policy.get("cohort_attempt_budget_multiplier", 4))))
|
|
attempts = hard_rejections = duplicate_rejections = soft_reviews = model_requests = replacement_attempts = 0
|
|
workers = self._effective_concurrency(cohort.concurrency)
|
|
peak_concurrency = 0
|
|
generation_records: list[dict[str, Any]] = []
|
|
while len(proposals) < cohort.cohort_size and attempts < total_budget:
|
|
batch_size = min(workers, total_budget - attempts, cohort.cohort_size - len(proposals) + workers - 1)
|
|
batch_inputs = []
|
|
for _ in range(batch_size):
|
|
attempts += 1
|
|
model_requests += 1
|
|
replacement_attempts += 1 if attempts > cohort.cohort_size else 0
|
|
slot_index = len(proposals) + 1
|
|
territory = mandate.opportunity_territories[(attempts - 1) % len(mandate.opportunity_territories)] if mandate.opportunity_territories else OpportunityTerritory.OPEN_CATEGORY
|
|
batch_inputs.append({"attempt": attempts, "slot_index": slot_index, "territory": str(territory)})
|
|
|
|
def generate_attempt(item: dict[str, Any]) -> dict[str, Any]:
|
|
context = {"hard_exclusions": mandate.hard_exclusions, "soft_exclusions": mandate.soft_exclusions, "territory": item["territory"], "brief": self.ideation_mandate_payload(mandate)}
|
|
payload, source = self._company_payload(cohort.mandate, ideation_index=item["attempt"], ideation_context=context)
|
|
return {**item, "payload": payload, "source": source}
|
|
|
|
for result in self._bounded_map(batch_inputs, generate_attempt, cohort.concurrency):
|
|
peak_concurrency = max(peak_concurrency, self._last_bounded_map_peak)
|
|
gate = self.novelty_gate(result["payload"], cohort, accepted_payloads, mandate, territory=result["territory"])
|
|
generation_records.append({"attempt": result["attempt"], "decision": gate["decision"], "title": result["payload"].get("title", ""), "territory": result["territory"]})
|
|
if gate["decision"] == NoveltyGateDecision.ACCEPT and len(proposals) < cohort.cohort_size:
|
|
proposal = self.generate_single_company(cohort.mandate, ideation_index=len(proposals) + 1, payload=result["payload"], source=result["source"])
|
|
generation_index = len(proposals) + 1
|
|
proposal.metadata = {**proposal.metadata, "cohort_id": cohort.cohort_id, "independent_generation_index": generation_index, "prior_ideas_visible": False, "search_territory": result["territory"], "novelty_gate": {k: v for k, v in gate.items() if k != "matched_thesis"}}
|
|
proposal.save(update_fields=["metadata", "updated_at"])
|
|
VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"generation_index": generation_index, "source_attempt": result["attempt"]})
|
|
proposals.append(proposal)
|
|
accepted_payloads.append(result["payload"])
|
|
continue
|
|
if gate["decision"] == NoveltyGateDecision.REGENERATE_HARD_EXCLUSION:
|
|
hard_rejections += 1
|
|
elif gate["decision"] == NoveltyGateDecision.REGENERATE_DUPLICATE:
|
|
duplicate_rejections += 1
|
|
elif gate["decision"] == NoveltyGateDecision.REVIEW_SOFT_EXCLUSION:
|
|
soft_reviews += 1
|
|
VentureGenerationRejection.objects.create(cohort=cohort, slot_index=result["slot_index"], attempt=result["attempt"], decision=gate["decision"], reason=gate["explanation"], candidate=result["payload"], matched_thesis=gate.get("matched_thesis"), similarity_score=float(gate.get("similarity_score", 0.0)), metadata={"territory": result["territory"], "classification": gate.get("classification"), "semantic_cluster_key": gate.get("semantic_cluster_key", "")})
|
|
if len(proposals) >= cohort.cohort_size:
|
|
break
|
|
failed_slots = max(0, cohort.cohort_size - len(proposals))
|
|
if failed_slots:
|
|
for missing in range(len(proposals) + 1, cohort.cohort_size + 1):
|
|
VentureGenerationRejection.objects.create(cohort=cohort, slot_index=missing, attempt=attempts, decision=NoveltyGateDecision.FAILED_IDEATION, reason="cohort-level attempt budget exhausted", metadata={"attempt_budget": total_budget})
|
|
cohort.status = "PROPOSALS_GENERATED" if len(proposals) == cohort.cohort_size else "INSUFFICIENT_ACCEPTED_PROPOSALS"
|
|
cohort.metrics = {**cohort.metrics, "requested_company_count": cohort.cohort_size, "generation_attempts": attempts, "cohort_attempt_budget": total_budget, "accepted_proposals": len(proposals), "replacement_attempts": replacement_attempts, "hard_exclusion_rejections": hard_rejections, "duplicate_rejections": duplicate_rejections, "soft_exclusion_reviews": soft_reviews, "failed_slots": failed_slots, "average_attempts_per_company": round(attempts / max(1, len(proposals)), 2), "generation_model_requests": model_requests, "proposal_generation_runtime_seconds": round(time.monotonic() - started, 2), "proposal_generation_peak_concurrency": peak_concurrency, "peak_concurrency": max(cohort.metrics.get("peak_concurrency", 0), peak_concurrency), "generation_records": generation_records[-50:]}
|
|
cohort.save(update_fields=["metrics", "status", "updated_at"])
|
|
return proposals
|
|
|
|
def research_cohort(self, cohort: VentureCohort) -> None:
|
|
started = time.monotonic()
|
|
|
|
def research(member_id: str) -> int:
|
|
member = VentureCohortMember.objects.select_related("proposal").get(id=member_id)
|
|
return len(self.conduct_market_research(member.proposal, depth="light").get("sources", []))
|
|
|
|
member_ids = [str(member.id) for member in cohort.members.order_by("created_at")]
|
|
source_counts = self._bounded_map(member_ids, research, cohort.concurrency)
|
|
peak_concurrency = self._last_bounded_map_peak
|
|
source_rejections = sum((member.proposal.metadata.get("research", {}) if isinstance(member.proposal.metadata, dict) else {}).get("source_rejection_count", 0) for member in cohort.members.select_related("proposal"))
|
|
cohort.metrics = {**cohort.metrics, "research_runtime_seconds": round(time.monotonic() - started, 2), "total_sources": sum(source_counts), "source_rejection_count": source_rejections, "public_research_queries": len(member_ids), "research_peak_concurrency": peak_concurrency, "peak_concurrency": max(cohort.metrics.get("peak_concurrency", 0), peak_concurrency)}
|
|
cohort.status = "RESEARCHED"
|
|
cohort.save(update_fields=["metrics", "status", "updated_at"])
|
|
|
|
def fingerprint_cohort(self, cohort: VentureCohort) -> list[VentureThesisFingerprint]:
|
|
return [self.fingerprint_proposal(member.proposal) for member in cohort.members.select_related("proposal")]
|
|
|
|
def cluster_theses(self, cohort: VentureCohort) -> list[PortfolioThesisCluster]:
|
|
clusters = []
|
|
for member in cohort.members.select_related("proposal"):
|
|
proposal = member.proposal
|
|
self.fingerprint_proposal(proposal)
|
|
match = self.match_portfolio_thesis(proposal)
|
|
thesis = match.get("matched_thesis")
|
|
if thesis is None:
|
|
thesis = self.create_portfolio_thesis_from_proposal(proposal, cohort)
|
|
match = {**match, "matched_thesis": thesis, "classification": ThesisMatchClassification.NEW_THESIS, "similarity_score": 1.0, "explanation": "Created a new portfolio thesis for this opportunity area."}
|
|
cluster, _ = PortfolioThesisCluster.objects.update_or_create(
|
|
cohort=cohort,
|
|
proposal=proposal,
|
|
defaults={
|
|
"portfolio_thesis": thesis,
|
|
"similarity_score": float(match["similarity_score"]),
|
|
"classification": match["classification"],
|
|
"explanation": match["explanation"],
|
|
"metadata": {"canonical_name": thesis.canonical_name},
|
|
},
|
|
)
|
|
clusters.append(cluster)
|
|
proposal.metadata = {**proposal.metadata, "portfolio_thesis_id": str(thesis.id), "portfolio_thesis": thesis.canonical_name, "thesis_match": {"classification": cluster.classification, "similarity_score": cluster.similarity_score, "explanation": cluster.explanation}}
|
|
proposal.save(update_fields=["metadata", "updated_at"])
|
|
return clusters
|
|
|
|
def analyze_collisions(self, cohort: VentureCohort) -> list[VentureCollision]:
|
|
collisions = []
|
|
proposals = [member.proposal for member in cohort.members.select_related("proposal")]
|
|
for a, b in combinations(proposals, 2):
|
|
result = self.classify_overlap(a, b)
|
|
collisions.append(VentureCollision.objects.create(cohort=cohort, company_a=a, company_b=b, classification=result["classification"], similarity_score=result["similarity_score"], explanation=result["explanation"], overlapping_dimensions=result["overlapping_dimensions"]))
|
|
return collisions
|
|
|
|
def run_individual_diligence_for_cohort(self, cohort: VentureCohort) -> None:
|
|
started = time.monotonic()
|
|
|
|
def diligence(member_id: str) -> None:
|
|
member = VentureCohortMember.objects.select_related("proposal").get(id=member_id)
|
|
proposal = member.proposal
|
|
self.board_review(proposal)
|
|
diligence = self.start_ic_diligence(proposal)
|
|
self.generate_ic_questions(diligence)
|
|
self.answer_questions(diligence)
|
|
self.red_team(diligence)
|
|
self.final_company_response(diligence)
|
|
self.score_and_decide(diligence)
|
|
self.assess_autonomous_operability(proposal)
|
|
gap = self.capability_analysis(proposal)
|
|
self.produce_investment_memo(diligence, gap)
|
|
member.child_graph_run = GraphRun.objects.create(execution_graph_version=cohort.graph_run.execution_graph_version if cohort.graph_run else None, status=GraphRunStatus.COMPLETE, metadata={"logical_child_company_run": True, "proposal_id": str(proposal.id), "cohort_id": cohort.cohort_id}) if cohort.graph_run else None
|
|
member.save(update_fields=["child_graph_run", "updated_at"])
|
|
|
|
member_ids = [str(member.id) for member in cohort.members.order_by("created_at")]
|
|
self._bounded_map(member_ids, diligence, cohort.concurrency)
|
|
peak_concurrency = self._last_bounded_map_peak
|
|
cohort.metrics = {**cohort.metrics, "individual_diligence_runtime_seconds": round(time.monotonic() - started, 2), "individual_diligence_peak_concurrency": peak_concurrency, "peak_concurrency": max(cohort.metrics.get("peak_concurrency", 0), peak_concurrency)}
|
|
cohort.status = "INDIVIDUAL_DILIGENCE_COMPLETE"
|
|
cohort.save(update_fields=["metrics", "status", "updated_at"])
|
|
|
|
def assess_autonomous_operability_for_cohort(self, cohort: VentureCohort) -> list[AutonomousOperabilityAssessment]:
|
|
assessments = [self.assess_autonomous_operability(member.proposal) for member in cohort.members.select_related("proposal")]
|
|
cohort.metrics = {**cohort.metrics, "autonomous_eligible_count": sum(1 for item in assessments if item.gate_result == AutonomousGateResult.AUTONOMOUS_ELIGIBLE), "assisted_only_count": sum(1 for item in assessments if item.gate_result == AutonomousGateResult.ASSISTED_ONLY)}
|
|
cohort.save(update_fields=["metrics", "updated_at"])
|
|
return assessments
|
|
|
|
def assess_autonomous_operability(self, proposal: CompanyProposal) -> AutonomousOperabilityAssessment:
|
|
text = " ".join([proposal.title, proposal.description, proposal.problem, proposal.target_customer, proposal.proposed_solution, proposal.business_model, proposal.pricing_hypothesis, proposal.acquisition_strategy, proposal.validation_plan, proposal.differentiation]).lower()
|
|
positive_terms = ["self-service", "automated", "monitor", "subscription", "api", "dashboard", "report", "digital", "crawl", "alert", "standardized", "recurring", "checkout", "plugin", "cli", "no-code"]
|
|
negative_causes = {
|
|
"SALES": ["sales call", "enterprise sales", "procurement", "relationship"],
|
|
"DOMAIN_EXPERTISE": ["expert", "clinical", "medical", "licensed", "engineer review"],
|
|
"CUSTOMER_TRUST": ["trust", "advisor", "consultant"],
|
|
"RELATIONSHIP_MANAGEMENT": ["account management", "relationship"],
|
|
"MANUAL_QA": ["manual qa", "manual review", "human review"],
|
|
"REGULATORY_SIGNOFF": ["regulatory sign-off", "compliance signoff", "approval"],
|
|
"CUSTOM_IMPLEMENTATION": ["custom implementation", "integration", "bespoke"],
|
|
"SUPPORT_ESCALATION": ["high-touch support", "white glove"],
|
|
"OFFLINE_ACTIVITY": ["onsite", "offline", "in person"],
|
|
"LEGAL_JUDGMENT": ["legal judgment", "legal advice", "lawyer"],
|
|
"NEGOTIATION": ["negotiation"],
|
|
"PERSONAL_BRAND": ["personal brand", "founder credibility"],
|
|
}
|
|
causes = [cause for cause, terms in negative_causes.items() if any(term in text for term in terms)]
|
|
score = 45 + min(30, sum(5 for term in positive_terms if term in text)) - min(45, len(causes) * 8)
|
|
ai = self.ai_leverage_score(proposal)
|
|
platform = self.platformization_potential(proposal)
|
|
score += 10 if ai >= 70 else 0
|
|
score += 10 if platform >= 70 else 0
|
|
score = max(0, min(100, score))
|
|
minutes = 5 if score >= 90 else 15 if score >= 80 else 30 if score >= 70 else 90 if score >= 50 else 240
|
|
dependency = FounderDependencyLevel.NONE if minutes <= 5 and not causes else FounderDependencyLevel.LOW if minutes <= 30 and len(causes) <= 1 else FounderDependencyLevel.MEDIUM if minutes <= 90 else FounderDependencyLevel.HIGH if minutes <= 240 else FounderDependencyLevel.CRITICAL
|
|
if score >= 80 and dependency in {FounderDependencyLevel.NONE, FounderDependencyLevel.LOW} and ai >= 70 and platform >= 70:
|
|
gate = AutonomousGateResult.AUTONOMOUS_ELIGIBLE
|
|
track = VentureTrack.AUTONOMOUS
|
|
elif score >= 70 and dependency in {FounderDependencyLevel.LOW, FounderDependencyLevel.MEDIUM}:
|
|
gate = AutonomousGateResult.AUTONOMOUS_BORDERLINE
|
|
track = VentureTrack.AUTONOMOUS
|
|
elif score >= 45:
|
|
gate = AutonomousGateResult.ASSISTED_ONLY
|
|
track = VentureTrack.ASSISTED
|
|
else:
|
|
gate = AutonomousGateResult.REJECT_OPERABILITY
|
|
track = VentureTrack.ASSISTED
|
|
loop = self.autonomous_operating_loop(proposal, gate)
|
|
contract = self.autonomous_fulfillment_contract(proposal)
|
|
assessment, _ = AutonomousOperabilityAssessment.objects.update_or_create(
|
|
proposal=proposal,
|
|
defaults={"venture_track": track, "gate_result": gate, "autonomous_operability_score": score, "commercial_score": self.commercial_score(proposal), "founder_dependency": dependency, "dependency_causes": causes, "minutes_per_week_human": minutes, "human_actions_required": contract["escalation"], "human_action_categories": ["LEGAL_SETUP", "SPEND_APPROVAL", "EXCEPTION_ESCALATION"] if minutes <= 30 else causes, "operating_loop": loop, "fulfillment_contract": contract, "validation_offer": self.validation_offer(proposal), "end_state_business_model": self.end_state_business_model(proposal), "structural_blockers": causes, "platform_blockers": self.platform_blockers(proposal), "component_scores": {"customer_acquisition_autonomy": loop["discover"]["score"], "onboarding_autonomy": loop["onboard"]["score"], "fulfillment_autonomy": loop["fulfill"]["score"], "support_autonomy": loop["support"]["score"], "billing_payment_autonomy": loop["checkout"]["score"], "quality_verification": loop["verify"]["score"], "repeatability": 90 if "recurring" in text or "subscription" in text else 65}, "rationale": f"{gate}: score {score}, dependency {dependency}, causes {', '.join(causes) or 'none'}."},
|
|
)
|
|
proposal.metadata = {**proposal.metadata, "venture_track": track, "autonomous_operability_score": score, "founder_dependency": dependency, "autonomous_gate_result": gate, "minutes_per_week_human": minutes, "validation_offer": assessment.validation_offer, "end_state_business_model": assessment.end_state_business_model}
|
|
proposal.save(update_fields=["metadata", "updated_at"])
|
|
self._artifact(proposal, proposal.mandate, "AUTONOMOUS_OPERABILITY_REPORT", "Autonomous Operability Report", {"score": score, "gate_result": gate, "founder_dependency": dependency, "dependency_causes": causes, "operating_loop": loop, "fulfillment_contract": contract, "validation_offer": assessment.validation_offer, "end_state_business_model": assessment.end_state_business_model, "platform_blockers": assessment.platform_blockers, "structural_blockers": assessment.structural_blockers}, self._readable_memo({"score": score, "gate_result": gate, "founder_dependency": dependency, "dependency_causes": causes, "operating_loop": loop}), "Autonomous Operability IC")
|
|
return assessment
|
|
|
|
def commercial_score(self, proposal: CompanyProposal) -> float:
|
|
diligence = proposal.ic_diligence.order_by("-created_at").first()
|
|
return float(diligence.decision.composite_score) if diligence and hasattr(diligence, "decision") else 0.0
|
|
|
|
def autonomous_operating_loop(self, proposal: CompanyProposal, gate: str) -> dict[str, dict[str, Any]]:
|
|
automated = gate in {AutonomousGateResult.AUTONOMOUS_ELIGIBLE, AutonomousGateResult.AUTONOMOUS_BORDERLINE}
|
|
state = "AUTONOMOUS" if automated else "HUMAN_ROUTINE_REQUIRED"
|
|
score = 90 if automated else 35
|
|
return {stage: {"capability": "AVAILABLE" if stage in {"discover", "checkout", "deliver", "measure"} else "PARTIAL", "operation": state, "score": score} for stage in ["discover", "qualify", "acquire", "checkout", "onboard", "fulfill", "verify", "deliver", "support", "measure", "retain_upsell"]}
|
|
|
|
def autonomous_fulfillment_contract(self, proposal: CompanyProposal) -> dict[str, Any]:
|
|
return {"customer_input": "Customer supplies a URL, file, repository, document, dataset, or narrow workflow input through a self-service intake.", "artifex_process": "Artifex agents run retrieval, analysis/transformation, structured QA, report generation, and delivery workflows.", "customer_output": proposal.proposed_solution, "quality_verification": "Automated checks validate schema completeness, source links, threshold scores, and known failure conditions before delivery.", "failure_recovery": "If verification fails, rerun once with stricter constraints; unresolved failures create exceptional human escalation.", "billing_event": "Payment is collected at checkout before one-off validation delivery or at subscription activation.", "support_model": "Agent-authored help, status updates, and retry guidance; human handles only exceptional safety/legal/account issues.", "escalation": ["material legal commitment", "irreversible spend", "security/safety exception", "verification repeatedly fails"]}
|
|
|
|
def validation_offer(self, proposal: CompanyProposal) -> dict[str, Any]:
|
|
return {"offer": "$49-$99 fixed-scope automated validation output", "price": "$49-$99", "proof": "willingness-to-pay for a standardized digital result before full product build", "aligned_with_end_state": True}
|
|
|
|
def end_state_business_model(self, proposal: CompanyProposal) -> dict[str, Any]:
|
|
return {"model": proposal.business_model, "likely_pricing": proposal.pricing_hypothesis, "recurring_path": "subscription, usage, or repeat purchase if validation demand repeats"}
|
|
|
|
def platform_blockers(self, proposal: CompanyProposal) -> list[str]:
|
|
return ["STRIPE_IMPLEMENTATION_MISSING", "SUBDOMAIN_DEPLOYMENT", "EMAIL_DELIVERY", "SUPPORT_INBOX"]
|
|
|
|
def portfolio_ic(self, cohort: VentureCohort) -> PortfolioICReview:
|
|
self.assess_autonomous_operability_for_cohort(cohort)
|
|
preliminary_rows = self.portfolio_ranking_rows(cohort)
|
|
preliminary_top_5 = preliminary_rows[:5]
|
|
before_coverage = {}
|
|
for row in preliminary_top_5:
|
|
proposal = CompanyProposal.objects.get(id=row["proposal_id"])
|
|
before_coverage[str(proposal.id)] = (proposal.metadata.get("research", {}) if isinstance(proposal.metadata, dict) else {}).get("coverage_ratio", 0.0)
|
|
deep_started = time.monotonic()
|
|
for row in preliminary_top_5:
|
|
proposal = CompanyProposal.objects.get(id=row["proposal_id"])
|
|
self.conduct_market_research(proposal, depth="deep")
|
|
diligence = proposal.ic_diligence.order_by("-created_at").first()
|
|
if diligence:
|
|
self.score_and_decide(diligence)
|
|
after_coverage = {}
|
|
for row in preliminary_top_5:
|
|
proposal = CompanyProposal.objects.get(id=row["proposal_id"])
|
|
after_coverage[str(proposal.id)] = (proposal.metadata.get("research", {}) if isinstance(proposal.metadata, dict) else {}).get("coverage_ratio", 0.0)
|
|
rows = self.portfolio_ranking_rows(cohort)
|
|
preliminary_rank_by_id = {row["proposal_id"]: index + 1 for index, row in enumerate(preliminary_rows)}
|
|
ranking_changes = [{"proposal_id": row["proposal_id"], "company": row["company"], "preliminary_rank": preliminary_rank_by_id.get(row["proposal_id"]), "final_rank": index + 1} for index, row in enumerate(rows) if preliminary_rank_by_id.get(row["proposal_id"]) != index + 1]
|
|
top_3 = [row for row in rows if row.get("autonomous_gate_result") == AutonomousGateResult.AUTONOMOUS_ELIGIBLE][:3]
|
|
top_3_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.is_top_3 = row["proposal_id"] in top_3_ids
|
|
member.portfolio_score = row["portfolio_score"]
|
|
member.save(update_fields=["rank", "is_top_3", "portfolio_score", "updated_at"])
|
|
row["rank"] = rank
|
|
concentration = self._portfolio_concentration(cohort)
|
|
review, _ = PortfolioICReview.objects.update_or_create(cohort=cohort, defaults={"rankings": rows, "top_3": top_3, "concentration": concentration, "metadata": {"no_funding": True, "preliminary_rankings": preliminary_rows, "preliminary_top_5": preliminary_top_5, "finalist_research_coverage_before": before_coverage, "finalist_research_coverage_after": after_coverage, "ranking_changes_after_deep_research": ranking_changes, "autonomous_top_3_shortfall": max(0, 3 - len(top_3))}})
|
|
cohort.metrics = {**cohort.metrics, "finalist_deep_research_count": len(preliminary_top_5), "finalist_deep_research_runtime_seconds": round(time.monotonic() - deep_started, 2), "finalist_research_coverage_before": before_coverage, "finalist_research_coverage_after": after_coverage, "ranking_changes_after_deep_research": ranking_changes}
|
|
cohort.status = "PORTFOLIO_IC_COMPLETE"
|
|
cohort.save(update_fields=["metrics", "status", "updated_at"])
|
|
return review
|
|
|
|
def portfolio_ranking_rows(self, cohort: VentureCohort) -> list[dict[str, Any]]:
|
|
rows = []
|
|
collision_risk = defaultdict(int)
|
|
for collision in cohort.collisions.exclude(classification=OverlapClassification.NONE):
|
|
collision_risk[str(collision.company_a_id)] += 1
|
|
collision_risk[str(collision.company_b_id)] += 1
|
|
for member in cohort.members.select_related("proposal"):
|
|
proposal = member.proposal
|
|
decision = proposal.ic_diligence.order_by("-created_at").first().decision
|
|
assessment = self.assess_autonomous_operability(proposal)
|
|
capability_burden = proposal.capability_requirements.filter(status=CapabilityStatus.MISSING).count()
|
|
concentration_penalty = self._generic_concentration_penalty(proposal)
|
|
ai_bonus = (decision.component_scores.get("AI Leverage", 0) * 0.08) + (decision.component_scores.get("Platformization Potential", 0) * 0.06)
|
|
autonomy_penalty = 35 if assessment.gate_result == AutonomousGateResult.ASSISTED_ONLY else 70 if assessment.gate_result == AutonomousGateResult.REJECT_OPERABILITY else 12 if assessment.gate_result == AutonomousGateResult.AUTONOMOUS_BORDERLINE else 0
|
|
score = round(decision.composite_score * 0.45 + assessment.autonomous_operability_score * 0.45 + decision.probability_500_within_30_days * 0.12 + ai_bonus - capability_burden * 1.5 - collision_risk[str(proposal.id)] * 2 - concentration_penalty - autonomy_penalty, 1)
|
|
rows.append({"proposal_id": str(proposal.id), "company": proposal.title, "thesis": proposal.pitch.get("One-line thesis", proposal.description), "ic_score": decision.composite_score, "commercial_score": assessment.commercial_score, "autonomy_score": assessment.autonomous_operability_score, "autonomous_gate_result": assessment.gate_result, "founder_dependency": assessment.founder_dependency, "minutes_per_week_human": assessment.minutes_per_week_human, "probability": decision.probability_500_within_30_days, "decision": decision.decision, "evidence_tier": decision.evidence_tier, "initial_tranche": str(decision.initial_tranche or "0"), "ai_leverage": decision.component_scores.get("AI Leverage", 0), "platformization_potential": decision.component_scores.get("Platformization Potential", 0), "portfolio_score": score, "capability_burden": capability_burden, "collision_risk": collision_risk[str(proposal.id)], "concentration_penalty": concentration_penalty})
|
|
rows.sort(key=lambda item: item["portfolio_score"], reverse=True)
|
|
return rows
|
|
|
|
def saturation_analysis(self, cohort: VentureCohort) -> list[dict[str, Any]]:
|
|
rows = []
|
|
by_thesis: dict[str, list[PortfolioThesisCluster]] = defaultdict(list)
|
|
for cluster in cohort.thesis_clusters.select_related("portfolio_thesis", "proposal"):
|
|
if cluster.portfolio_thesis_id:
|
|
by_thesis[str(cluster.portfolio_thesis_id)].append(cluster)
|
|
for clusters in by_thesis.values():
|
|
thesis = clusters[0].portfolio_thesis
|
|
scores = []
|
|
classifications = Counter(cluster.classification for cluster in clusters)
|
|
best_proposal = None
|
|
for cluster in clusters:
|
|
diligence = cluster.proposal.ic_diligence.order_by("-created_at").first()
|
|
if diligence and hasattr(diligence, "decision"):
|
|
scores.append(float(diligence.decision.composite_score))
|
|
if best_proposal is None or diligence.decision.composite_score > best_proposal[1]:
|
|
best_proposal = (cluster.proposal, float(diligence.decision.composite_score))
|
|
best_score = max(scores) if scores else 0.0
|
|
spread = round(max(scores) - min(scores), 1) if len(scores) > 1 else 0.0
|
|
proposal_count = len(clusters)
|
|
near_count = classifications.get(ThesisMatchClassification.NEAR_DUPLICATE, 0) + classifications.get(ThesisMatchClassification.DUPLICATE, 0)
|
|
if proposal_count >= 3 and near_count >= 2 and spread <= 15:
|
|
recommendation = PortfolioThesisStatus.SATURATED
|
|
rationale = "Multiple semantically similar proposals with limited score spread; more ideation is unlikely to add information without a new wedge."
|
|
elif proposal_count >= 2 and best_score >= 60 and spread <= 10:
|
|
recommendation = PortfolioThesisStatus.SATURATED
|
|
rationale = "A differentiated winner appears to have emerged; additional near-term ideation should move elsewhere."
|
|
elif best_score >= 55:
|
|
recommendation = PortfolioThesisStatus.ACTIVE_CANDIDATE
|
|
rationale = "Strong enough to remain active but not yet saturated."
|
|
else:
|
|
recommendation = PortfolioThesisStatus.EXPLORED
|
|
rationale = "Explored without enough evidence or score strength to prioritize."
|
|
analysis, _ = PortfolioSaturationAnalysis.objects.update_or_create(
|
|
cohort=cohort,
|
|
portfolio_thesis=thesis,
|
|
defaults={"proposal_count": proposal_count, "best_ic_score": best_score, "score_spread": spread, "status_recommendation": recommendation, "rationale": rationale, "metadata": {"classification_counts": dict(classifications), "best_company": best_proposal[0].title if best_proposal else ""}},
|
|
)
|
|
rows.append({"thesis": thesis.canonical_name, "proposal_count": analysis.proposal_count, "best_ic_score": analysis.best_ic_score, "score_spread": analysis.score_spread, "status_recommendation": analysis.status_recommendation, "rationale": analysis.rationale})
|
|
if hasattr(cohort, "portfolio_review"):
|
|
review = cohort.portfolio_review
|
|
review.metadata = {**review.metadata, "saturation_analysis": rows}
|
|
review.save(update_fields=["metadata", "updated_at"])
|
|
cohort.metadata = {**cohort.metadata, "saturation_analysis": rows}
|
|
cohort.save(update_fields=["metadata", "updated_at"])
|
|
return rows
|
|
|
|
def update_thesis_registry(self, cohort: VentureCohort) -> list[dict[str, Any]]:
|
|
updates = []
|
|
for analysis in cohort.saturation_analyses.select_related("portfolio_thesis"):
|
|
thesis = analysis.portfolio_thesis
|
|
clusters = list(cohort.thesis_clusters.filter(portfolio_thesis=thesis).select_related("proposal"))
|
|
best = None
|
|
for cluster in clusters:
|
|
diligence = cluster.proposal.ic_diligence.order_by("-created_at").first()
|
|
if diligence and hasattr(diligence, "decision") and (best is None or diligence.decision.composite_score > best[1]):
|
|
best = (cluster.proposal, float(diligence.decision.composite_score))
|
|
thesis.proposal_count = thesis.proposal_clusters.count()
|
|
thesis.best_ic_score = max(float(thesis.best_ic_score or 0.0), analysis.best_ic_score)
|
|
if best and (thesis.best_company is None or best[1] >= thesis.best_ic_score):
|
|
thesis.best_company = best[0]
|
|
thesis.first_seen_cohort = thesis.first_seen_cohort or cohort
|
|
thesis.last_seen_cohort = cohort
|
|
thesis.status = analysis.status_recommendation
|
|
best_assessment = self.assess_autonomous_operability(best[0]) if best else None
|
|
if best_assessment:
|
|
thesis.venture_track = best_assessment.venture_track
|
|
thesis.best_autonomous_operability_score = max(float(thesis.best_autonomous_operability_score or 0.0), best_assessment.autonomous_operability_score)
|
|
thesis.best_founder_dependency = best_assessment.founder_dependency
|
|
thesis.autonomous_candidate_status = AutonomousCandidateStatus.AUTONOMOUS_CANDIDATE if best_assessment.gate_result == AutonomousGateResult.AUTONOMOUS_ELIGIBLE else AutonomousCandidateStatus.ASSISTED_CANDIDATE if best_assessment.gate_result == AutonomousGateResult.ASSISTED_ONLY else AutonomousCandidateStatus.AUTONOMOUS_REJECTED if best_assessment.gate_result == AutonomousGateResult.REJECT_OPERABILITY else AutonomousCandidateStatus.NOT_ASSESSED
|
|
thesis.metadata = {**thesis.metadata, "last_saturation_rationale": analysis.rationale, "last_reopen_reason": thesis.metadata.get("last_reopen_reason", "")}
|
|
thesis.save(update_fields=["proposal_count", "best_ic_score", "best_company", "first_seen_cohort", "last_seen_cohort", "status", "venture_track", "best_autonomous_operability_score", "best_founder_dependency", "autonomous_candidate_status", "metadata", "updated_at"])
|
|
updates.append({"thesis": thesis.canonical_name, "status": thesis.status, "proposal_count": thesis.proposal_count, "best_ic_score": thesis.best_ic_score, "venture_track": thesis.venture_track, "best_autonomous_operability_score": thesis.best_autonomous_operability_score, "best_founder_dependency": thesis.best_founder_dependency, "autonomous_candidate_status": thesis.autonomous_candidate_status})
|
|
cohort.metadata = {**cohort.metadata, "thesis_registry_after": self.registry_snapshot()}
|
|
cohort.save(update_fields=["metadata", "updated_at"])
|
|
return updates
|
|
|
|
def aggregate_capability_demand(self, cohort: VentureCohort, *, top_3_only: bool = False) -> list[dict[str, Any]]:
|
|
members = cohort.members.filter(is_top_3=True) if top_3_only else cohort.members.all()
|
|
proposals = [member.proposal for member in members.select_related("proposal")]
|
|
by_capability: dict[str, list[CompanyCapabilityRequirement]] = defaultdict(list)
|
|
for proposal in proposals:
|
|
for req in proposal.capability_requirements.all():
|
|
by_capability[req.category].append(req)
|
|
stage_order = {CapabilityPriority.BEFORE_VALIDATION: 0, CapabilityPriority.BEFORE_FIRST_CUSTOMER: 1, CapabilityPriority.BEFORE_SCALING: 2}
|
|
results = []
|
|
for capability, reqs in by_capability.items():
|
|
earliest = sorted([req.priority for req in reqs], key=lambda item: stage_order[item])[0]
|
|
statuses = Counter(req.status for req in reqs)
|
|
top_3_count = sum(1 for req in reqs if req.proposal.cohort_memberships.filter(cohort=cohort, is_top_3=True).exists())
|
|
priority_score = len(reqs) * 10 + top_3_count * 8 + (12 if earliest == CapabilityPriority.BEFORE_VALIDATION else 6 if earliest == CapabilityPriority.BEFORE_FIRST_CUSTOMER else 2) + statuses.get(CapabilityStatus.MISSING, 0) * 4
|
|
row = {"capability": capability, "count": len(reqs), "percentage": round((len(reqs) / max(1, len(proposals))) * 100, 1), "earliest_stage": earliest, "companies": [req.proposal.title for req in reqs], "status_distribution": dict(statuses), "top_3_count": top_3_count, "priority_score": priority_score}
|
|
if not top_3_only:
|
|
VentureCapabilityDemand.objects.update_or_create(cohort=cohort, capability=capability, defaults={k: v for k, v in row.items() if k != "capability"})
|
|
results.append(row)
|
|
results.sort(key=lambda item: item["priority_score"], reverse=True)
|
|
if hasattr(cohort, "portfolio_review"):
|
|
review = cohort.portfolio_review
|
|
if top_3_only:
|
|
review.top_3_capability_gaps = results[:10]
|
|
else:
|
|
review.capability_demand = results
|
|
review.recommended_build_priorities = [item["capability"] for item in results if CapabilityStatus.MISSING in item["status_distribution"]][:5]
|
|
review.save(update_fields=["capability_demand", "top_3_capability_gaps", "recommended_build_priorities", "updated_at"])
|
|
return results
|
|
|
|
def produce_cohort_report(self, cohort: VentureCohort) -> VentureArtifact:
|
|
review = cohort.portfolio_review
|
|
ideation_mandate = getattr(cohort, "ideation_mandate", None)
|
|
clusters = [{"company": cluster.proposal.title, "portfolio_thesis": cluster.portfolio_thesis.canonical_name if cluster.portfolio_thesis else "", "classification": cluster.classification, "similarity_score": cluster.similarity_score, "explanation": cluster.explanation} for cluster in cohort.thesis_clusters.select_related("proposal", "portfolio_thesis").order_by("created_at")]
|
|
autonomy = [self.autonomy_report_row(member.proposal) for member in cohort.members.select_related("proposal").order_by("rank", "created_at")]
|
|
generation_rejections = [{"slot_index": rejection.slot_index, "attempt": rejection.attempt, "decision": rejection.decision, "reason": rejection.reason, "candidate_title": rejection.candidate.get("title", "") if isinstance(rejection.candidate, dict) else "", "similarity_score": rejection.similarity_score} for rejection in cohort.generation_rejections.order_by("slot_index", "attempt")]
|
|
saturation = [{"thesis": analysis.portfolio_thesis.canonical_name, "proposal_count": analysis.proposal_count, "best_ic_score": analysis.best_ic_score, "score_spread": analysis.score_spread, "status_recommendation": analysis.status_recommendation, "rationale": analysis.rationale} for analysis in cohort.saturation_analyses.select_related("portfolio_thesis").order_by("-proposal_count")]
|
|
accepted_count = cohort.members.count()
|
|
content = {"cohort_id": cohort.cohort_id, "mandate": cohort.mandate.objective, "accepted_count": accepted_count, "requested_count": cohort.cohort_size, "venture_track": VentureTrack.AUTONOMOUS, "autonomous_operability_report": autonomy, "assisted_only_companies": [row for row in autonomy if row.get("gate_result") == AutonomousGateResult.ASSISTED_ONLY], "ideation_mandate": self.ideation_mandate_payload(ideation_mandate) if ideation_mandate else {}, "hard_exclusions": ideation_mandate.hard_exclusions if ideation_mandate else [], "soft_exclusions": ideation_mandate.soft_exclusions if ideation_mandate else [], "search_territories": ideation_mandate.opportunity_territories if ideation_mandate else [], "generation_rejections": generation_rejections, "thesis_registry_before": (ideation_mandate.registry_snapshot.get("theses", []) if ideation_mandate else []), "thesis_registry_after": cohort.metadata.get("thesis_registry_after", []), "thesis_clusters": clusters, "saturation_analysis": saturation, "idea_diversity_metrics": self.diversity_metrics(cohort), "runtime": cohort.metrics, "total_spend": 0, "customer_outreach": "none", "rankings": review.rankings, "top_3": review.top_3, "collisions": self._collision_summary(cohort), "portfolio_concentration": review.concentration, "capability_demand": review.capability_demand, "top_3_capability_gaps": review.top_3_capability_gaps, "recommended_build_priorities": review.recommended_build_priorities}
|
|
readable = self._readable_cohort_report(content)
|
|
cohort.status = "COMPLETE" if accepted_count == cohort.cohort_size else "PARTIAL_COMPLETE"
|
|
cohort.save(update_fields=["status", "updated_at"])
|
|
return VentureArtifact.objects.create(mandate=cohort.mandate, graph_run=cohort.graph_run, artifact_type="VENTURE_DISCOVERY_COHORT_REPORT", name="Venture Discovery Cohort Report", content=content, readable=readable, generated_by="Portfolio IC")
|
|
|
|
def autonomy_report_row(self, proposal: CompanyProposal) -> dict[str, Any]:
|
|
assessment = self.assess_autonomous_operability(proposal)
|
|
return {"company": proposal.title, "venture_track": assessment.venture_track, "gate_result": assessment.gate_result, "commercial_score": assessment.commercial_score, "autonomous_operability_score": assessment.autonomous_operability_score, "founder_dependency": assessment.founder_dependency, "dependency_causes": assessment.dependency_causes, "minutes_per_week_human": assessment.minutes_per_week_human, "human_actions_required": assessment.human_actions_required, "operating_loop": assessment.operating_loop, "fulfillment_contract": assessment.fulfillment_contract, "validation_offer": assessment.validation_offer, "end_state_business_model": assessment.end_state_business_model, "structural_blockers": assessment.structural_blockers, "platform_blockers": assessment.platform_blockers, "component_scores": assessment.component_scores, "rationale": assessment.rationale}
|
|
|
|
def calibrate_probability(self, proposal: CompanyProposal, *, raw_probability: float) -> dict[str, Any]:
|
|
tier = proposal.evidence_tier or EvidenceTier.TIER_0_THESIS
|
|
ceiling = float(EVIDENCE_CEILINGS.get(tier, 35))
|
|
adjusted = min(float(raw_probability), ceiling)
|
|
return {"raw_probability": float(raw_probability), "evidence_adjusted_probability": adjusted, "evidence_ceiling": ceiling, "evidence_tier": tier, "explanation": f"{tier} caps P($500/30d) at {ceiling}%; IC uses {adjusted}%."}
|
|
|
|
def _bounded_map(self, items: list[Any] | range, worker: Any, concurrency: int) -> list[Any]:
|
|
self._last_bounded_map_peak = 0
|
|
workers = self._effective_concurrency(concurrency)
|
|
if workers <= 1:
|
|
results = [worker(item) for item in items]
|
|
self._last_bounded_map_peak = 1 if results else 0
|
|
return results
|
|
results: list[tuple[int, Any]] = []
|
|
active = 0
|
|
active_lock = threading.Lock()
|
|
|
|
def tracked_worker(item: Any) -> Any:
|
|
nonlocal active
|
|
with active_lock:
|
|
active += 1
|
|
self._last_bounded_map_peak = max(self._last_bounded_map_peak, active)
|
|
try:
|
|
return self._threaded_worker(worker, item)
|
|
finally:
|
|
with active_lock:
|
|
active -= 1
|
|
|
|
with ThreadPoolExecutor(max_workers=workers) as executor:
|
|
futures = {executor.submit(tracked_worker, item): index for index, item in enumerate(items)}
|
|
for future in as_completed(futures):
|
|
results.append((futures[future], future.result()))
|
|
results.sort(key=lambda item: item[0])
|
|
return [result for _, result in results]
|
|
|
|
def _threaded_worker(self, worker: Any, item: Any) -> Any:
|
|
close_old_connections()
|
|
try:
|
|
return worker(item)
|
|
finally:
|
|
close_old_connections()
|
|
|
|
def _effective_concurrency(self, concurrency: int) -> int:
|
|
requested = max(1, int(concurrency or 1))
|
|
if connection.vendor == "sqlite" and connection.settings_dict.get("NAME") == ":memory:":
|
|
return 1
|
|
return requested
|
|
|
|
def validate_identity_content(self, proposal: CompanyProposal, content: Any) -> dict[str, Any]:
|
|
text = json.dumps(content, default=str).lower()
|
|
checks = {
|
|
"company_identity": self._token_overlap(proposal.title, text) > 0,
|
|
"icp": self._token_overlap(proposal.target_customer, text) >= 1,
|
|
"problem": self._token_overlap(proposal.problem, text) >= 1,
|
|
"offer": self._token_overlap(proposal.proposed_solution, text) >= 1,
|
|
"business_model": self._token_overlap(proposal.business_model, text) >= 1,
|
|
}
|
|
passed = sum(1 for value in checks.values() if value) >= 3 and checks["company_identity"]
|
|
result = {"passed": passed, "checks": checks, "warning": "" if passed else "identity_consistency_warning"}
|
|
return result
|
|
|
|
def fingerprint_proposal(self, proposal: CompanyProposal) -> VentureThesisFingerprint:
|
|
try:
|
|
existing = proposal.fingerprint
|
|
except ObjectDoesNotExist:
|
|
existing = None
|
|
data = self._fingerprint_data(proposal)
|
|
digest = hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
|
|
fields = {**data, "fingerprint_hash": digest, "metadata": {"deterministic": True}}
|
|
if existing:
|
|
for key, value in fields.items():
|
|
setattr(existing, key, value)
|
|
existing.save(update_fields=[*fields.keys(), "updated_at"])
|
|
return existing
|
|
return VentureThesisFingerprint.objects.create(proposal=proposal, **fields)
|
|
|
|
def classify_overlap(self, a: CompanyProposal, b: CompanyProposal) -> dict[str, Any]:
|
|
fa = self.fingerprint_proposal(a)
|
|
fb = self.fingerprint_proposal(b)
|
|
dimensions = ["industry", "business_model", "primary_distribution_channel", "price_band", "time_to_first_cash_band", "service_software_hybrid", "regulatory_dependency"]
|
|
overlap = [dim for dim in dimensions if getattr(fa, dim) and getattr(fa, dim) == getattr(fb, dim)]
|
|
text_score = self._jaccard(f"{fa.icp} {fa.problem} {fa.offer}", f"{fb.icp} {fb.problem} {fb.offer}")
|
|
score = round((len(overlap) / len(dimensions)) * 0.55 + text_score * 0.45, 2)
|
|
if score >= 0.82:
|
|
classification = OverlapClassification.DUPLICATE
|
|
elif score >= 0.62:
|
|
classification = OverlapClassification.NEAR_DUPLICATE
|
|
elif getattr(fa, "industry") == getattr(fb, "industry") and getattr(fa, "icp") == getattr(fb, "icp"):
|
|
classification = OverlapClassification.COMPETITIVE
|
|
elif getattr(fa, "industry") == getattr(fb, "industry") or getattr(fa, "business_model") == getattr(fb, "business_model"):
|
|
classification = OverlapClassification.ADJACENT
|
|
else:
|
|
classification = OverlapClassification.NONE
|
|
return {"classification": classification, "similarity_score": score, "overlapping_dimensions": overlap, "explanation": f"Overlap {overlap}; text similarity {text_score:.2f}."}
|
|
|
|
def seed_dogfood_thesis_registry(self) -> None:
|
|
seeds = [
|
|
("AI RFP response automation for B2B SaaS", "AI-assisted RFP/proposal drafting and response automation for B2B SaaS sales teams.", "AI wrapper platforms", PortfolioThesisStatus.SATURATED),
|
|
("AI churn/retention intelligence for SaaS", "AI analysis of SaaS customer health data to predict churn and generate retention playbooks.", "AI-enabled services", PortfolioThesisStatus.SATURATED),
|
|
("Shopify compliance/audit", "Shopify or ecommerce audit/compliance products and services.", "SMB automation", PortfolioThesisStatus.EXPLORED),
|
|
("Freelancer contract risk scanning", "Contract risk review and clause scanning for freelancers and solo operators.", "Data/document automation", PortfolioThesisStatus.ACTIVE_CANDIDATE),
|
|
]
|
|
for name, thesis, category, status in seeds:
|
|
existing = PortfolioThesis.objects.filter(canonical_name=name).first()
|
|
historical = self.historical_thesis_stats(name)
|
|
historical_metadata = {key: value for key, value in historical.items() if key != "best_company"}
|
|
historical_metadata["best_company"] = historical["best_company"].title if historical["best_company"] else ""
|
|
defaults = {"concise_thesis": thesis, "category": category, "status": status, "metadata": {"dogfood_seed": True, "historical": historical_metadata}}
|
|
if historical["proposal_count"]:
|
|
defaults.update({"proposal_count": historical["proposal_count"], "best_ic_score": historical["best_ic_score"], "best_company": historical["best_company"]})
|
|
if existing is None:
|
|
PortfolioThesis.objects.create(canonical_name=name, fingerprint=hashlib.sha256(name.lower().encode()).hexdigest(), **defaults)
|
|
else:
|
|
if existing.metadata.get("last_reopen_reason"):
|
|
defaults.pop("status", None)
|
|
for key, value in defaults.items():
|
|
setattr(existing, key, value)
|
|
existing.save(update_fields=[*defaults.keys(), "updated_at"])
|
|
|
|
def historical_thesis_stats(self, canonical_name: str) -> dict[str, Any]:
|
|
proposals = [proposal for proposal in CompanyProposal.objects.all() if self.canonical_thesis_name(self.payload_from_proposal(proposal)) == canonical_name]
|
|
best_score = 0.0
|
|
best_company = None
|
|
for proposal in proposals:
|
|
diligence = proposal.ic_diligence.order_by("-created_at").first()
|
|
if diligence and hasattr(diligence, "decision") and diligence.decision.composite_score > best_score:
|
|
best_score = float(diligence.decision.composite_score)
|
|
best_company = proposal
|
|
return {"proposal_count": len(proposals), "best_ic_score": best_score, "best_company": best_company}
|
|
|
|
def registry_snapshot(self) -> list[dict[str, Any]]:
|
|
return [{"id": str(thesis.id), "canonical_name": thesis.canonical_name, "status": thesis.status, "proposal_count": thesis.proposal_count, "best_ic_score": thesis.best_ic_score, "best_company": thesis.best_company.title if thesis.best_company else "", "last_seen_cohort": thesis.last_seen_cohort.cohort_id if thesis.last_seen_cohort else "", "category": thesis.category} for thesis in PortfolioThesis.objects.order_by("canonical_name")]
|
|
|
|
def reopen_portfolio_thesis(self, thesis: PortfolioThesis, *, reason: str, status: str = PortfolioThesisStatus.ACTIVE_CANDIDATE) -> PortfolioThesis:
|
|
thesis.status = status
|
|
thesis.metadata = {**thesis.metadata, "last_reopen_reason": reason, "reopened_at": timezone.now().isoformat()}
|
|
thesis.save(update_fields=["status", "metadata", "updated_at"])
|
|
return thesis
|
|
|
|
def allocate_territories(self, size: int) -> list[str]:
|
|
base = [item.value for item in DEFAULT_TERRITORY_ALLOCATION]
|
|
return [base[index % len(base)] for index in range(max(1, size))]
|
|
|
|
def ideation_mandate_payload(self, mandate: CohortIdeationMandate) -> dict[str, Any]:
|
|
return {"objective": mandate.objective, "desired_opportunity_classes": mandate.desired_opportunity_classes, "hard_exclusions": mandate.hard_exclusions, "soft_exclusions": mandate.soft_exclusions, "opportunity_territories": mandate.opportunity_territories, "portfolio_gaps": mandate.portfolio_gaps, "diversity_preferences": mandate.diversity_preferences, "registry_snapshot": mandate.registry_snapshot}
|
|
|
|
def ideation_brief_text(self, review: dict[str, Any], territories: list[str]) -> str:
|
|
return "\n".join([
|
|
"COHORT IDEATION BRIEF",
|
|
"Mandate: intentionally search new venture territory while preserving independent ideation.",
|
|
"Desired opportunity classes: AI wrapper platforms; agentic workflow products; vertical copilots; AI-enabled services; developer / AI infrastructure; intelligence products; automation products; service -> platform paths; recurring revenue; owned-compute leverage.",
|
|
"HARD EXCLUSIONS: " + "; ".join(DEFAULT_HARD_EXCLUSIONS),
|
|
"SOFT EXCLUSIONS: " + "; ".join(DEFAULT_SOFT_EXCLUSIONS),
|
|
"SATURATED THESIS AREAS: " + "; ".join(row["canonical_name"] for row in review.get("saturated_thesis_areas", [])),
|
|
"ACTIVE CANDIDATES: " + "; ".join(row["canonical_name"] for row in review.get("active_candidates", [])),
|
|
"UNDEREXPLORED TERRITORIES: developer infrastructure; intelligence monitoring; data/document automation; non-RFP enterprise workflows; non-churn SaaS intelligence.",
|
|
"SEARCH TERRITORY ALLOCATION: " + "; ".join(territories),
|
|
])
|
|
|
|
def novelty_gate(self, payload: dict[str, Any], cohort: VentureCohort, accepted_payloads: list[dict[str, Any]], mandate: CohortIdeationMandate, *, territory: str) -> dict[str, Any]:
|
|
hard = self.hard_exclusion_match(payload, mandate.hard_exclusions)
|
|
if hard:
|
|
return {"decision": NoveltyGateDecision.REGENERATE_HARD_EXCLUSION, "classification": ThesisMatchClassification.DUPLICATE, "similarity_score": 1.0, "explanation": f"Hard-excluded thesis area: {hard}", "territory": territory}
|
|
soft = self.soft_exclusion_match(payload, mandate.soft_exclusions)
|
|
if soft and not self.has_soft_exception(payload):
|
|
return {"decision": NoveltyGateDecision.REVIEW_SOFT_EXCLUSION, "classification": ThesisMatchClassification.ADJACENT, "similarity_score": 0.55, "explanation": f"Soft-excluded area lacks material differentiation: {soft}", "territory": territory}
|
|
registry_match = self.match_payload_to_registry(payload)
|
|
thesis = registry_match.get("matched_thesis")
|
|
if thesis and thesis.status == PortfolioThesisStatus.SATURATED and registry_match["classification"] in {ThesisMatchClassification.NEAR_DUPLICATE, ThesisMatchClassification.DUPLICATE, ThesisMatchClassification.COMPETITIVE}:
|
|
return {"decision": NoveltyGateDecision.REGENERATE_DUPLICATE, **registry_match, "explanation": "Matched saturated thesis: " + registry_match["explanation"], "territory": territory}
|
|
cohort_match = self.current_cohort_match(payload, accepted_payloads)
|
|
if cohort_match["classification"] == ThesisMatchClassification.DUPLICATE:
|
|
return {"decision": NoveltyGateDecision.REGENERATE_DUPLICATE, **cohort_match, "territory": territory}
|
|
if cohort_match["classification"] == ThesisMatchClassification.NEAR_DUPLICATE:
|
|
cluster_count = sum(1 for item in accepted_payloads if self.semantic_cluster_key(item) == cohort_match.get("semantic_cluster_key"))
|
|
if cluster_count >= int(DEFAULT_DUPLICATE_POLICY["max_near_duplicate_per_thesis_per_cohort"]) + 1:
|
|
return {"decision": NoveltyGateDecision.REGENERATE_DUPLICATE, **cohort_match, "territory": territory}
|
|
if cohort_match["classification"] == ThesisMatchClassification.COMPETITIVE:
|
|
canonical = self.canonical_thesis_name(payload)
|
|
competitive_count = sum(1 for item in accepted_payloads if self.canonical_thesis_name(item) == canonical)
|
|
if competitive_count >= int(DEFAULT_DUPLICATE_POLICY["max_competitive_per_thesis_per_cohort"]):
|
|
return {"decision": NoveltyGateDecision.REGENERATE_DUPLICATE, **cohort_match, "territory": territory}
|
|
return {"decision": NoveltyGateDecision.ACCEPT, "classification": registry_match.get("classification", ThesisMatchClassification.NEW_THESIS), "similarity_score": registry_match.get("similarity_score", 0.0), "explanation": registry_match.get("explanation", "Accepted as novel enough for this cohort."), "territory": territory, "matched_thesis": thesis.id if thesis else None, "soft_exception": soft or ""}
|
|
|
|
def hard_exclusion_match(self, payload: dict[str, Any], exclusions: list[str]) -> str:
|
|
text = self.payload_text(payload)
|
|
canonical = self.canonical_thesis_name(payload).lower()
|
|
if "rfp" in text or "request for proposal" in text or "proposal drafting" in text:
|
|
return "AI RFP response drafting / proposal automation"
|
|
if "churn" in text or "retention playbook" in text or "save play" in text:
|
|
return "SaaS churn prediction / retention playbook generation"
|
|
for exclusion in exclusions:
|
|
if self._jaccard(exclusion, canonical + " " + text) >= 0.25:
|
|
return exclusion
|
|
return ""
|
|
|
|
def soft_exclusion_match(self, payload: dict[str, Any], exclusions: list[str]) -> str:
|
|
text = self.payload_text(payload)
|
|
checks = [("generic contract scanners", ["contract", "scanner"]), ("Shopify audit products", ["shopify", "audit"]), ("generic AI content generators", ["content", "generator"]), ("generic productized consulting", ["consulting"]), ("generic one-off audits", ["audit"]), ("generic chatbot wrappers", ["chatbot"])]
|
|
for label, words in checks:
|
|
if label in exclusions and all(word in text for word in words):
|
|
return label
|
|
return ""
|
|
|
|
def has_soft_exception(self, payload: dict[str, Any]) -> bool:
|
|
rationale = str(payload.get("exception_rationale", "") or payload.get("differentiation", "")).lower()
|
|
return any(word in rationale for word in ["vertical", "specific", "proprietary", "workflow", "recurring", "data", "platform", "compliance", "regulated"])
|
|
|
|
def match_portfolio_thesis(self, proposal: CompanyProposal) -> dict[str, Any]:
|
|
return self.match_payload_to_registry(self.payload_from_proposal(proposal))
|
|
|
|
def match_payload_to_registry(self, payload: dict[str, Any]) -> dict[str, Any]:
|
|
canonical = self.canonical_thesis_name(payload)
|
|
best = None
|
|
for thesis in PortfolioThesis.objects.all():
|
|
score = 1.0 if thesis.canonical_name == canonical else self._jaccard(self.payload_text(payload), " ".join([thesis.canonical_name, thesis.concise_thesis, thesis.icp, thesis.problem, thesis.offer, thesis.business_model, thesis.primary_channel]))
|
|
if best is None or score > best[0]:
|
|
best = (score, thesis)
|
|
if best is None or best[0] < 0.18:
|
|
return {"matched_thesis": None, "similarity_score": 0.0, "classification": ThesisMatchClassification.NEW_THESIS, "explanation": "No existing portfolio thesis was similar enough."}
|
|
score, thesis = best
|
|
classification = ThesisMatchClassification.DUPLICATE if score >= 0.9 else ThesisMatchClassification.NEAR_DUPLICATE if score >= 0.62 else ThesisMatchClassification.COMPETITIVE if score >= 0.38 else ThesisMatchClassification.ADJACENT
|
|
return {"matched_thesis": thesis, "similarity_score": round(score, 2), "classification": classification, "explanation": f"Matched {thesis.canonical_name} at similarity {score:.2f}."}
|
|
|
|
def current_cohort_match(self, payload: dict[str, Any], accepted_payloads: list[dict[str, Any]]) -> dict[str, Any]:
|
|
canonical = self.canonical_thesis_name(payload)
|
|
cluster_key = self.semantic_cluster_key(payload)
|
|
best_score = 0.0
|
|
best_name = ""
|
|
for accepted in accepted_payloads:
|
|
accepted_name = self.canonical_thesis_name(accepted)
|
|
if str(accepted.get("title", "")).strip().lower() == str(payload.get("title", "")).strip().lower():
|
|
return {"matched_thesis": None, "similarity_score": 1.0, "classification": ThesisMatchClassification.DUPLICATE, "explanation": f"Current cohort already accepted exact title {accepted.get('title', '')}.", "semantic_cluster_key": cluster_key}
|
|
accepted_cluster_key = self.semantic_cluster_key(accepted)
|
|
score = 0.86 if accepted_cluster_key == cluster_key else self._jaccard(self.payload_text(payload), self.payload_text(accepted))
|
|
if score > best_score:
|
|
best_score = score
|
|
best_name = accepted_name
|
|
classification = ThesisMatchClassification.NEW_THESIS
|
|
if best_score >= 0.78:
|
|
classification = ThesisMatchClassification.NEAR_DUPLICATE
|
|
elif best_score >= 0.55:
|
|
classification = ThesisMatchClassification.COMPETITIVE
|
|
elif best_score >= 0.18:
|
|
classification = ThesisMatchClassification.ADJACENT
|
|
return {"matched_thesis": None, "similarity_score": round(best_score, 2), "classification": classification, "explanation": f"Current cohort nearest thesis {best_name or 'none'} at similarity {best_score:.2f}.", "semantic_cluster_key": cluster_key}
|
|
|
|
def create_portfolio_thesis_from_proposal(self, proposal: CompanyProposal, cohort: VentureCohort) -> PortfolioThesis:
|
|
fingerprint = self.fingerprint_proposal(proposal)
|
|
canonical = self.canonical_thesis_name(self.payload_from_proposal(proposal))
|
|
thesis, _ = PortfolioThesis.objects.update_or_create(
|
|
canonical_name=canonical,
|
|
defaults={"concise_thesis": proposal.pitch.get("One-line thesis", proposal.description), "category": str(proposal.metadata.get("search_territory", "")), "industry": fingerprint.industry, "icp": fingerprint.icp, "problem": fingerprint.problem, "offer": fingerprint.offer, "business_model": fingerprint.business_model, "primary_channel": fingerprint.primary_distribution_channel, "fingerprint": fingerprint.fingerprint_hash, "status": PortfolioThesisStatus.NEW, "first_seen_cohort": cohort, "last_seen_cohort": cohort, "metadata": {"created_from_proposal": str(proposal.id)}},
|
|
)
|
|
return thesis
|
|
|
|
def canonical_thesis_name(self, payload: dict[str, Any]) -> str:
|
|
text = self.payload_text(payload)
|
|
if "rfp" in text or "request for proposal" in text or "proposal drafting" in text:
|
|
return "AI RFP response automation for B2B SaaS"
|
|
if "churn" in text or "retention" in text or "save play" in text:
|
|
return "AI churn/retention intelligence for SaaS"
|
|
if "contract" in text and ("freelance" in text or "freelancer" in text):
|
|
return "Freelancer contract risk scanning"
|
|
if "shopify" in text and "audit" in text:
|
|
return "Shopify compliance/audit"
|
|
semantic = self.semantic_cluster_key(payload)
|
|
if semantic == "municipal_permit_compliance":
|
|
return "Municipal permit compliance automation"
|
|
if semantic == "legacy_modernization_triage":
|
|
return "Legacy modernization triage copilot"
|
|
title = str(payload.get("title", "Untitled thesis")).strip()
|
|
return re.sub(r"\s+", " ", title)[:255]
|
|
|
|
def semantic_cluster_key(self, payload: dict[str, Any]) -> str:
|
|
text = self.payload_text(payload)
|
|
if ("permit" in text or "permitdoc" in text) and any(word in text for word in ["compliance", "pre-check", "precheck", "municipal", "contractor"]):
|
|
return "municipal_permit_compliance"
|
|
if "legacy" in text and any(word in text for word in ["modernization", "modernisation", "triage", "copilot", "agent"]):
|
|
return "legacy_modernization_triage"
|
|
if "vendor" in text and "onboarding" in text and "compliance" in text:
|
|
return "vendor_onboarding_compliance"
|
|
title = str(payload.get("title", "")).lower()
|
|
tokens = [token for token in re.findall(r"[a-z0-9]{4,}", title) if token not in {"agent", "copilot", "checker", "monitor", "platform", "automation", "workflow", "service", "precheck", "check"}]
|
|
return "_".join(tokens[:4]) or self._fingerprint(payload)[:12]
|
|
|
|
def payload_text(self, payload: dict[str, Any]) -> str:
|
|
return " ".join(str(payload.get(key, "")) for key in ["title", "one_line_thesis", "description", "problem", "target_customer", "proposed_solution", "business_model", "pricing_hypothesis", "acquisition_strategy", "validation_plan", "differentiation"]).lower()
|
|
|
|
def payload_from_proposal(self, proposal: CompanyProposal) -> dict[str, Any]:
|
|
return {"title": proposal.title, "one_line_thesis": proposal.pitch.get("One-line thesis", ""), "description": proposal.description, "problem": proposal.problem, "target_customer": proposal.target_customer, "proposed_solution": proposal.proposed_solution, "business_model": proposal.business_model, "pricing_hypothesis": proposal.pricing_hypothesis, "acquisition_strategy": proposal.acquisition_strategy, "validation_plan": proposal.validation_plan, "differentiation": proposal.differentiation}
|
|
|
|
def diversity_metrics(self, cohort: VentureCohort) -> dict[str, Any]:
|
|
clusters = [cluster for cluster in cohort.thesis_clusters.select_related("portfolio_thesis", "proposal")]
|
|
proposals = [member.proposal for member in cohort.members.select_related("proposal")]
|
|
fingerprint_rows = [self.fingerprint_values(proposal) for proposal in proposals]
|
|
industries = Counter(row["industry"] for row in fingerprint_rows)
|
|
business_models = Counter(row["business_model"] for row in fingerprint_rows)
|
|
channels = Counter(row["primary_distribution_channel"] for row in fingerprint_rows)
|
|
thesis_counts = Counter(cluster.portfolio_thesis.canonical_name if cluster.portfolio_thesis else "unclustered" for cluster in clusters)
|
|
icp_groups = Counter(row["industry"] + ":" + row["price_band"] for row in fingerprint_rows)
|
|
total = max(1, len(proposals))
|
|
return {"unique_thesis_clusters": len(thesis_counts), "total_companies": len(proposals), "unique_thesis_clusters_ratio": round(len(thesis_counts) / total, 2), "unique_industries": len(industries), "unique_icp_groups": len(icp_groups), "unique_business_models": len(business_models), "unique_primary_channels": len(channels), "largest_thesis_cluster_size": max(thesis_counts.values()) if thesis_counts else 0, "largest_industry_concentration": max(industries.values()) if industries else 0, "largest_business_model_concentration": max(business_models.values()) if business_models else 0, "previous_qwen_baseline": {"companies": 10, "rfp_variants": 4, "churn_retention_variants": 5}, "thesis_counts": dict(thesis_counts), "industry_counts": dict(industries), "business_model_counts": dict(business_models), "primary_channel_counts": dict(channels)}
|
|
|
|
def fingerprint_values(self, proposal: CompanyProposal) -> dict[str, Any]:
|
|
try:
|
|
fingerprint = proposal.fingerprint
|
|
return {"industry": fingerprint.industry, "business_model": fingerprint.business_model, "primary_distribution_channel": fingerprint.primary_distribution_channel, "price_band": fingerprint.price_band}
|
|
except ObjectDoesNotExist:
|
|
return self._fingerprint_data(proposal)
|
|
|
|
def _company_payload(self, mandate: CompanyMandate, *, ideation_index: int | None = None, ideation_context: dict[str, Any] | None = None) -> tuple[dict[str, Any], str]:
|
|
if self.router is not None:
|
|
try:
|
|
slot = f" Independent cohort slot: {ideation_index}. Do not use or imitate other cohort ideas; no other ideas are visible." if ideation_index else ""
|
|
context = ideation_context or {}
|
|
response = self.router.complete(ModelRequestContract(purpose=ModelCapability.PLANNING, model_hint=self.ideation_model_hint, prompt="Generate exactly ONE startup idea for an AUTONOMOUS Venture Discovery cohort. Return a single JSON object, not a list. The business must be operable by Artifex with <=30 minutes/week routine human involvement. Prefer archetypes: " + json.dumps(AUTONOMOUS_ARCHETYPES) + ". Require concrete digital operating loop: discover lead/user -> qualify -> acquire -> checkout -> onboard -> fulfill -> verify -> deliver -> support -> measure -> retain/upsell. Include validation_offer and end_state_business_model as separate fields. Include fulfillment_contract with customer_input, artifex_process, customer_output, quality_verification, failure_recovery, billing_event, support_model, escalation. Avoid enterprise procurement, founder-led calls, bespoke consulting, legal/medical judgment, offline fulfillment, and custom implementation. Respect no spend and no outreach in V0. Keep the $50 to $500 in 30 days mandate. Prefer AI-native businesses where Artifex can deliver most value using agents/local inference, but do not allow 'uses AI' to substitute for real customer pain. DO NOT propose companies substantially equivalent to these hard-excluded thesis areas, including semantic variants: " + json.dumps(context.get("hard_exclusions", [])) + ". Soft exclusions require material differentiation and an explicit exception rationale: " + json.dumps(context.get("soft_exclusions", [])) + ". Search territory for this slot: " + str(context.get("territory", "OPEN_CATEGORY")) + ". Include title, one_line_thesis, description, problem, target_customer, proposed_solution, business_model, pricing_hypothesis, acquisition_strategy, validation_plan, capital_requested, time_to_first_dollar_estimate, expected_margin, build_complexity, market_evidence, differentiation, major_risks, confidence, minutes_per_week_human, human_actions_required, and optional exception_rationale." + slot + " Mandate: " + json.dumps({"objective": mandate.objective, "constraints": mandate.constraints, "optimization_targets": mandate.optimization_targets, "ai_native_policy": AI_NATIVE_POLICY, "ideation_brief": context.get("brief", {})})))
|
|
parsed = extract_json_object(response.content)
|
|
if isinstance(parsed, dict) and parsed.get("title"):
|
|
return self._normalize_payload(parsed), self.ideation_model_hint
|
|
except Exception:
|
|
pass
|
|
payload = self._fallback_company_payload()
|
|
payload["market_evidence"] = [{"type": "fallback_hypothesis", "source": "deterministic_fallback", "summary": "No Sol/web research was used; this is an internal hypothesis for test/resilience only.", "fallback_evidence": True}]
|
|
payload["confidence"] = 0.52
|
|
return payload, "deterministic_fallback"
|
|
|
|
def _fallback_company_payload(self) -> dict[str, Any]:
|
|
return {"title": "LaunchLens", "one_line_thesis": "A productized validation audit helps solo builders decide whether an AI-enabled microbusiness is worth pursuing before they spend weeks building.", "description": "LaunchLens sells a concise validation and launch-readiness report for one microbusiness idea.", "problem": "Solo technical founders often overbuild AI products before proving willingness to pay.", "target_customer": "Solo technical founders and small service operators considering an AI-assisted microbusiness.", "proposed_solution": "A fixed-scope paid audit that evaluates ICP, first-dollar route, validation gates, build plan, risks, and Artifex capability gaps.", "business_model": "Productized service first, with optional later software tooling if demand is proven.", "pricing_hypothesis": "$49-$99 per audit, with a higher-touch $250 implementation planning upsell after validation.", "acquisition_strategy": "Compliant founder-community posts, personal network asks, and content showing anonymized example audits after outreach is approved.", "validation_plan": "Before any build, seek 5 credible target-customer responses or 1 willingness-to-pay signal through compliant channels once approved.", "capital_requested": "50", "time_to_first_dollar_estimate": "3-10 days after outreach is approved", "expected_margin": "70-85% gross margin after manual delivery time", "build_complexity": "LOW", "market_evidence": [], "differentiation": "Combines venture IC-style diligence with Artifex's software/agent execution awareness and explicit capability-gap reporting.", "major_risks": ["Demand may be consulting-like and hard to differentiate.", "Manual distribution may not produce urgent buyers.", "No V0 customer evidence exists yet."], "confidence": 0.52}
|
|
|
|
def _normalize_payload(self, payload: dict[str, Any]) -> dict[str, Any]:
|
|
fallback = self._fallback_company_payload()
|
|
normalized = {key: payload.get(key, value) for key, value in fallback.items()}
|
|
normalized["market_evidence"] = self._as_list(normalized.get("market_evidence"))
|
|
normalized["major_risks"] = self._as_list(normalized.get("major_risks"))
|
|
if payload.get("exception_rationale"):
|
|
normalized["exception_rationale"] = payload["exception_rationale"]
|
|
for key in ["validation_offer", "end_state_business_model", "fulfillment_contract", "minutes_per_week_human", "human_actions_required"]:
|
|
if key in payload:
|
|
normalized[key] = payload[key]
|
|
return normalized
|
|
|
|
def _normalize_research(self, payload: dict[str, Any], required: list[str]) -> dict[str, Any]:
|
|
sources = []
|
|
for source in self._as_list(payload.get("sources"))[:12]:
|
|
if isinstance(source, dict) and source.get("url"):
|
|
sources.append({"type": "public_web", "url": str(source["url"]), "title": str(source.get("title", "")), "category": str(source.get("category", "")), "summary": str(source.get("summary", "")), "fallback_evidence": False})
|
|
coverage = {category: bool(dict(payload.get("coverage", {})).get(category)) for category in required}
|
|
return {"coverage": coverage, "sources": sources, "findings": dict(payload.get("findings", {})), "unverified_categories": []}
|
|
|
|
def _searxng_sources(self, proposal: CompanyProposal, required: list[str], *, depth: str = "light") -> list[dict[str, Any]]:
|
|
client = self.search_client or SearxngSearchClient.from_resources()
|
|
self._last_search_diagnostics = {"queries": [], "errors": []}
|
|
if client is None:
|
|
self._last_search_diagnostics["errors"].append("no active searxng resource")
|
|
return []
|
|
sources = []
|
|
limit = 5 if depth == "deep" else 3
|
|
query_terms = {
|
|
"competitors": "competitors alternatives",
|
|
"pricing": "pricing cost",
|
|
"customer_pain": "customer pain problem forum",
|
|
"market_alternatives": "alternatives tools services",
|
|
"regulatory_platform_risks": "regulatory platform risk compliance",
|
|
}
|
|
for category in required:
|
|
query = f"{proposal.title} {proposal.problem[:120]} {proposal.target_customer} {query_terms.get(category, category)}"
|
|
try:
|
|
results = self._search_client_results(client, query, category=category, limit=limit)
|
|
if not results:
|
|
fallback_query = f"{proposal.title} {query_terms.get(category, category)}"
|
|
results = self._search_client_results(client, fallback_query, category=category, limit=limit)
|
|
if not results:
|
|
broad_query = f"{proposal.target_customer} {query_terms.get(category, category)}"
|
|
results = self._search_client_results(client, broad_query, category=category, limit=limit)
|
|
sources.extend(results)
|
|
except Exception as exc:
|
|
self._last_search_diagnostics["errors"].append(f"{category}: {exc}")
|
|
continue
|
|
return sources[:25 if depth == "deep" else 15]
|
|
|
|
def _search_client_results(self, client: Any, query: str, *, category: str, limit: int) -> list[dict[str, Any]]:
|
|
if hasattr(client, "search_payload"):
|
|
payload = client.search_payload(query)
|
|
raw_results = payload.get("results", []) if isinstance(payload, dict) else []
|
|
unresponsive = payload.get("unresponsive_engines", []) if isinstance(payload, dict) else []
|
|
self._last_search_diagnostics["queries"].append({"category": category, "query": query, "result_count": len(raw_results), "unresponsive_engines": unresponsive})
|
|
return [
|
|
{"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}
|
|
for item in raw_results[: limit * 4]
|
|
if isinstance(item, dict) and item.get("url")
|
|
]
|
|
results = client.search(query, category=category, limit=limit)
|
|
self._last_search_diagnostics["queries"].append({"category": category, "query": query, "result_count": len(results)})
|
|
return results
|
|
|
|
def _page_corpus(self, search_sources: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
if self.page_fetcher is None:
|
|
return []
|
|
try:
|
|
return self.page_fetcher.fetch_many(search_sources, max_pages=8)
|
|
except Exception:
|
|
return []
|
|
|
|
def _merge_search_sources(self, research: dict[str, Any], search_sources: list[dict[str, Any]], required: list[str]) -> dict[str, Any]:
|
|
seen = {source.get("url") for source in research.get("sources", [])}
|
|
merged_sources = [*research.get("sources", [])]
|
|
for source in search_sources:
|
|
if source.get("url") in seen:
|
|
continue
|
|
seen.add(source.get("url"))
|
|
merged_sources.append(source)
|
|
coverage = {category: bool(dict(research.get("coverage", {})).get(category)) for category in required}
|
|
for source in merged_sources:
|
|
category = str(source.get("category", ""))
|
|
if category in coverage and source.get("url"):
|
|
coverage[category] = True
|
|
return {**research, "coverage": coverage, "sources": merged_sources, "research_available": True, "search_provider": "searxng"}
|
|
|
|
def filter_research_sources(self, proposal: CompanyProposal, sources: list[dict[str, Any]], required: list[str]) -> dict[str, list[dict[str, Any]]]:
|
|
accepted = []
|
|
rejected = []
|
|
seen = set()
|
|
for source in sources:
|
|
if not isinstance(source, dict):
|
|
continue
|
|
url = str(source.get("url", "")).strip()
|
|
if not url or url in seen:
|
|
continue
|
|
seen.add(url)
|
|
category = str(source.get("category", ""))
|
|
quality = self.source_quality(proposal, source, required)
|
|
if not quality["accepted"]:
|
|
rejected.append({**source, "quality": quality["quality"], "rejection_reason": quality["reason"]})
|
|
continue
|
|
accepted.append({**source, "quality": quality["quality"], "relevance_score": quality["score"]})
|
|
return {"accepted_sources": accepted, "rejected_sources": rejected}
|
|
|
|
def source_quality(self, proposal: CompanyProposal, source: dict[str, Any], required: list[str]) -> dict[str, Any]:
|
|
category = str(source.get("category", ""))
|
|
url = str(source.get("url", "")).lower()
|
|
source_text = " ".join(str(source.get(key, "")) for key in ["title", "summary", "content", "snippet", "url"]).lower()
|
|
if category not in required:
|
|
return {"accepted": False, "quality": "weak", "score": 0.0, "reason": "unknown research category"}
|
|
if any(noisy in url for noisy in ["webcache", "translate.google", "pinterest", "facebook.com", "instagram.com", "x.com/intent", "archive.org", "web.archive.org"]):
|
|
return {"accepted": False, "quality": "weak", "score": 0.0, "reason": "archive/social/noisy source"}
|
|
company_text = " ".join([proposal.title, proposal.problem, proposal.target_customer, proposal.proposed_solution]).lower()
|
|
category_terms = {
|
|
"competitors": "competitor competitors alternative alternatives vendor software platform service",
|
|
"pricing": "pricing price cost plan subscription fee rates",
|
|
"customer_pain": "problem pain challenge complaint forum reddit customer",
|
|
"market_alternatives": "alternative alternatives tools services products market solution",
|
|
"regulatory_platform_risks": "regulatory compliance risk policy legal platform permit",
|
|
}
|
|
company_score = self._jaccard(company_text, source_text)
|
|
category_score = self._jaccard(category_terms.get(category, category), source_text)
|
|
company_overlap = self._token_overlap(company_text, source_text)
|
|
category_overlap = self._token_overlap(category_terms.get(category, category), source_text)
|
|
if category.replace("_", " ") in source_text:
|
|
category_score = max(category_score, 0.5)
|
|
score = round(company_score * 0.55 + category_score * 0.25 + min(company_overlap, 5) * 0.03 + min(category_overlap, 4) * 0.025, 2)
|
|
if company_overlap >= 2 and category_overlap >= 1:
|
|
return {"accepted": True, "quality": "strong" if score >= 0.13 or category_overlap >= 2 else "weak", "score": score, "reason": "relevant proposal and category overlap"}
|
|
if score < 0.08:
|
|
return {"accepted": False, "quality": "weak", "score": score, "reason": "insufficient semantic relevance"}
|
|
if score < 0.16:
|
|
return {"accepted": True, "quality": "weak", "score": score, "reason": "accepted as weak context only"}
|
|
return {"accepted": True, "quality": "strong", "score": score, "reason": "strong relevant source"}
|
|
|
|
def _missing_research(self, required: list[str], reason: str) -> dict[str, Any]:
|
|
return {"coverage": {category: False for category in required}, "sources": [], "findings": {}, "unverified_categories": required, "research_available": False, "failure": reason}
|
|
|
|
def _as_list(self, value: Any) -> list[Any]:
|
|
if isinstance(value, list):
|
|
return value
|
|
if value in (None, ""):
|
|
return []
|
|
return [value]
|
|
|
|
def _money(self, value: Any) -> Decimal:
|
|
match = re.search(r"\d+(?:\.\d+)?", str(value))
|
|
if match is None:
|
|
return Decimal("50")
|
|
return min(Decimal(match.group(0)), Decimal("50"))
|
|
|
|
def _token_overlap(self, source: str, target_text: str) -> int:
|
|
tokens = {token for token in re.findall(r"[a-z0-9]{4,}", source.lower()) if token not in {"with", "that", "from", "this", "service", "business", "model", "customer", "customers"}}
|
|
return len([token for token in tokens if token in target_text])
|
|
|
|
def _jaccard(self, a: str, b: str) -> float:
|
|
left = {token for token in re.findall(r"[a-z0-9]{4,}", a.lower())}
|
|
right = {token for token in re.findall(r"[a-z0-9]{4,}", b.lower())}
|
|
if not left or not right:
|
|
return 0.0
|
|
return len(left & right) / len(left | right)
|
|
|
|
def _fingerprint_data(self, proposal: CompanyProposal) -> dict[str, Any]:
|
|
text = " ".join([proposal.title, proposal.description, proposal.problem, proposal.target_customer, proposal.proposed_solution, proposal.business_model, proposal.acquisition_strategy]).lower()
|
|
industry = "shopify/ecommerce" if "shopify" in text or "ecommerce" in text else "developer tools" if "developer" in text or "api" in text else "b2b services" if "b2b" in text else "general business"
|
|
model = "productized service" if "service" in proposal.business_model.lower() else "saas" if "saas" in proposal.business_model.lower() else "hybrid"
|
|
channel = "outbound/community" if any(word in text for word in ["outbound", "community", "posts", "network"]) else "marketplace" if "marketplace" in text or "fiverr" in text else "content/seo" if "seo" in text or "content" in text else "direct"
|
|
price = self._money(proposal.pricing_hypothesis)
|
|
price_band = "under_100" if price < 100 else "100_500" if price <= 500 else "over_500"
|
|
first_cash = "under_7_days" if any(token in proposal.time_to_first_dollar_estimate.lower() for token in ["3", "7", "week"]) else "under_30_days"
|
|
regulatory = "high" if any(word in text for word in ["legal", "compliance", "regulatory", "permit", "policy"]) else "medium" if "platform" in text else "low"
|
|
return {"industry": industry, "icp": proposal.target_customer[:500], "problem": proposal.problem[:500], "offer": proposal.proposed_solution[:500], "business_model": model, "primary_distribution_channel": channel, "price_band": price_band, "time_to_first_cash_band": first_cash, "required_capability_set": [item.category for item in proposal.capability_requirements.all()] or ["outbound sales", "CRM", "payments"], "geography_dependency": "local" if any(word in text for word in ["local", "city", "metro", "permit"]) else "none", "regulatory_dependency": regulatory, "online_offline": "online" if any(word in text for word in ["shopify", "saas", "api", "online", "web"]) else "mixed", "service_software_hybrid": model}
|
|
|
|
def _confidence(self, value: Any) -> float:
|
|
labels = {"low": 0.35, "medium": 0.55, "moderate": 0.55, "high": 0.75}
|
|
lowered = str(value).strip().lower()
|
|
if lowered in labels:
|
|
return labels[lowered]
|
|
match = re.search(r"\d+(?:\.\d+)?", lowered)
|
|
if match is None:
|
|
return 0.55
|
|
number = float(match.group(0))
|
|
return min(1.0, number / 100 if number > 1 else number)
|
|
|
|
def _evidence_scores(self, proposal: CompanyProposal) -> dict[str, int]:
|
|
research = proposal.metadata.get("research", {}) if isinstance(proposal.metadata, dict) else {}
|
|
coverage_ratio = float(research.get("coverage_ratio", 0.0) or 0.0)
|
|
unverified = set(research.get("unverified_categories", []))
|
|
evidence_count = len([item for item in self._as_list(proposal.market_evidence) if isinstance(item, dict) and item.get("url")])
|
|
low_build = str(proposal.build_complexity).upper() in {"LOW", "LOW-MEDIUM"}
|
|
service_model = "service" in proposal.business_model.lower()
|
|
margin_numbers = [int(value) for value in re.findall(r"\d+", proposal.expected_margin)]
|
|
margin = max(margin_numbers) if margin_numbers else 50
|
|
demand = min(85, 25 + int(coverage_ratio * 40) + min(evidence_count * 4, 20))
|
|
pricing = 70 if "pricing" not in unverified else 45
|
|
pain = 72 if "customer_pain" not in unverified else 35
|
|
competition = 68 if "competitors" not in unverified and "market_alternatives" not in unverified else 38
|
|
risk = 72 if "regulatory_platform_risks" not in unverified else 34
|
|
distribution = 60 if service_model else 42
|
|
build = 78 if low_build else 42
|
|
gross_margin = max(35, min(85, margin))
|
|
capital = 82 if proposal.capital_requested <= Decimal("50") else 30
|
|
validation = 78 if "5 credible" in proposal.validation_plan or "willingness" in proposal.validation_plan.lower() else 42
|
|
market_size = 62 if coverage_ratio >= 0.6 else 44
|
|
defensibility = 42 + (10 if service_model else 0) + (8 if coverage_ratio >= 0.8 else 0)
|
|
ai_leverage = self.ai_leverage_score(proposal)
|
|
platformization = self.platformization_potential(proposal)
|
|
probability = round((demand * 0.22 + pricing * 0.12 + pain * 0.16 + distribution * 0.14 + build * 0.1 + capital * 0.1 + risk * 0.16), 0)
|
|
autonomy = 55
|
|
try:
|
|
autonomy = int(proposal.autonomous_assessment.autonomous_operability_score)
|
|
except ObjectDoesNotExist:
|
|
pass
|
|
return {"Demand Evidence": demand, "Time-to-First-Dollar Attractiveness": 76 if service_model else 50, "Capital Efficiency": capital, "Validation Affordability": validation, "Gross Margin Potential": gross_margin, "Distribution Feasibility": distribution, "Build Simplicity": build, "Defensibility": min(75, defensibility), "Market Opportunity": market_size, "Competitive Position": competition, "Risk Manageability": risk, "AI Leverage": ai_leverage, "Platformization Potential": platformization, "Autonomous Operability": autonomy, "Probability of Reaching $500": int(max(20, min(80, probability)))}
|
|
|
|
def ai_leverage_score(self, proposal: CompanyProposal) -> int:
|
|
text = " ".join([proposal.title, proposal.description, proposal.problem, proposal.proposed_solution, proposal.business_model, proposal.differentiation]).lower()
|
|
score = 20
|
|
if any(term in text for term in ["ai", "agent", "model", "copilot", "automation", "inference", "llm"]):
|
|
score += 25
|
|
if any(term in text for term in ["monitor", "synthesis", "alerts", "intelligence", "workflow", "document", "evaluation", "deployment", "security", "data transformation"]):
|
|
score += 18
|
|
if any(term in text for term in ["recurring", "monthly", "subscription", "platform", "software", "api"]):
|
|
score += 15
|
|
if "manual" in text and not any(term in text for term in ["agent", "automation", "software", "platform"]):
|
|
score -= 15
|
|
if any(term in text for term in ["generic audit", "emergency fix", "one-time consulting"]):
|
|
score -= 12
|
|
return max(0, min(100, score))
|
|
|
|
def platformization_potential(self, proposal: CompanyProposal) -> int:
|
|
text = " ".join([proposal.title, proposal.description, proposal.proposed_solution, proposal.business_model, proposal.acquisition_strategy, proposal.validation_plan]).lower()
|
|
score = 25
|
|
if any(term in text for term in ["platform", "software", "saas", "api", "dashboard", "monitor", "alerts", "workflow"]):
|
|
score += 25
|
|
if any(term in text for term in ["repeatable", "template", "standardized", "recurring", "monthly", "subscription"]):
|
|
score += 20
|
|
if any(term in text for term in ["data", "history", "knowledge", "benchmark", "evaluation", "repository"]):
|
|
score += 12
|
|
if any(term in text for term in ["one-time", "emergency", "manual only", "concierge"]):
|
|
score -= 12
|
|
return max(0, min(100, score))
|
|
|
|
def _pitch(self, payload: dict[str, Any], *, fallback: bool) -> dict[str, Any]:
|
|
pitch = {"Company name": payload["title"], "One-line thesis": payload["one_line_thesis"], "Problem": payload["problem"], "ICP": payload["target_customer"], "Why now": "AI tooling lowers build cost, increasing the risk that founders overbuild before validating demand.", "Product / service": payload["proposed_solution"], "Business model": payload["business_model"], "Pricing": payload["pricing_hypothesis"], "Route to first customer": payload["acquisition_strategy"], "Validation plan": payload["validation_plan"], "$50 capital allocation proposal": {"initial": "$10 only after approval", "reserved": "$40 held until evidence gate", "v0_spend": "$0"}, "Time to first dollar": payload["time_to_first_dollar_estimate"], "Path to $500 net cash": "Sell 6-10 fixed-scope audits at $49-$99 while keeping delivery manual and using sunk Artifex compute.", "Competition": "Generic startup consultants, founder communities, AI business idea tools, and DIY validation templates. Public competitor research is unverified unless web research is configured.", "Differentiation": payload["differentiation"], "Build requirements": ["report template", "intake form", "manual analysis workflow", "optional landing page after validation approval"], "Distribution requirements": ["compliant outreach plan", "community/content channels", "CRM-lite tracking before first customers"], "Risks": payload["major_risks"], "What would falsify the thesis": "No willingness-to-pay signal at $49-$99 or fewer than 5 credible target-customer responses after approved compliant validation.", "Confidence": payload["confidence"], "Evidence caveat": "Deterministic fallback evidence only; not equivalent to model-generated or public web research." if fallback else "Generated by configured ideation model; public web research only included if sources are present."}
|
|
return pitch
|
|
|
|
def _questions_from_pitch(self, pitch: dict[str, Any]) -> list[dict[str, str]]:
|
|
return [{"category": "demand", "question": f"What evidence supports demand for {pitch['Company name']} among the stated ICP?"}, {"category": "urgency", "question": "Why will this customer pay now instead of using free templates or advice?"}, {"category": "distribution", "question": "How do you reach the first 10 customers without spam or fake traction?"}, {"category": "validation", "question": "Can this be validated before building the full product?"}, {"category": "business_model", "question": "Why a productized service first instead of SaaS?"}, {"category": "falsification", "question": "What would falsify the thesis within the $50 and 30-day mandate?"}, {"category": "economics", "question": "What happens if acquisition cost or manual delivery time is 3x the estimate?"}, {"category": "competition", "question": "What is the main competitive threat and why is this worth funding over selling an existing Artifex capability?"}]
|
|
|
|
def _answer_question(self, question: ICQuestion) -> dict[str, Any]:
|
|
evidence = [{"source": "internal_reasoning", "summary": "No customer outreach, spend, or fabricated evidence used.", "fallback_evidence": True}]
|
|
if question.category in {"demand", "competition"} and not self.web_research_available:
|
|
evidence.append({"source": "capability_gap", "summary": "Public web research unavailable; demand/competition claims remain uncertain.", "fallback_evidence": True})
|
|
answers = {"demand": "Demand is not proven. The strongest V0 claim is that the problem is plausible and cheap to test, not that demand exists.", "urgency": "The buyer pays only if the report saves them build time or prevents wasted spend; urgency is weakest before a concrete launch decision.", "distribution": "After approval, use targeted compliant posts/conversations and track responses manually; V0 performs no outreach.", "validation": "Yes. The paid diagnostic can be validated with responses and willingness-to-pay before software build.", "business_model": "Service first reduces build risk and can reach first cash faster than SaaS; software should follow only if repeated demand appears.", "falsification": "Failure to collect credible responses or willingness-to-pay within the mandate falsifies near-term viability.", "economics": "If acquisition or delivery is 3x harder, the company should stop or raise price before building tooling.", "competition": "Main threat is generic consulting/free templates. The reason to fund this over selling raw Artifex capability is packaging a buyer-specific outcome."}
|
|
return {"answer": answers.get(question.category, "The assumption remains uncertain and must be tested before spend."), "evidence": evidence, "uncertainty": "High until public research and customer evidence are available.", "pitch_changes": {"confidence_adjustment": "reduced/held due missing external evidence"} if question.category in {"demand", "competition"} else {}}
|
|
|
|
def _capability_requirements(self, proposal: CompanyProposal) -> list[dict[str, Any]]:
|
|
research = proposal.metadata.get("research", {}) if isinstance(proposal.metadata, dict) else {}
|
|
web_status = CapabilityStatus.MISSING if not self.web_research_available else CapabilityStatus.AVAILABLE if not research.get("unverified_categories") else CapabilityStatus.PARTIAL
|
|
try:
|
|
structural = proposal.autonomous_assessment.structural_blockers
|
|
except ObjectDoesNotExist:
|
|
structural = []
|
|
return [
|
|
{"category": "Company Brain", "status": CapabilityStatus.PARTIAL, "rationale": "Venture reasoning exists in V0 but is not a persistent operating brain.", "priority": CapabilityPriority.BEFORE_SCALING, "evidence": {}},
|
|
{"category": "Board", "status": CapabilityStatus.AVAILABLE, "rationale": "Structured CEO/CTO/CFO/CRO/Independent Director review exists for V0.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {}},
|
|
{"category": "IC", "status": CapabilityStatus.AVAILABLE, "rationale": "Bounded IC diligence, questions, scoring, and decision vocabulary exist.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {}},
|
|
{"category": "WEB_MARKET_RESEARCH", "status": web_status, "rationale": "Bounded source-linked web research exists only when configured and category coverage is complete.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {"web_research_available": self.web_research_available, **research}},
|
|
{"category": "software build", "status": CapabilityStatus.AVAILABLE, "rationale": "Task execution, coding, review, tests, and graph runtime exist.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
|
|
{"category": "frontend design", "status": CapabilityStatus.AVAILABLE, "rationale": "Frontend agents and Django UI path exist.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
|
|
{"category": "deployment", "status": CapabilityStatus.PARTIAL, "rationale": "Validation can use SUBDOMAIN_DEPLOYMENT under a shared parent domain; dedicated domains are a traction-stage upgrade.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {"deployment_model": "SUBDOMAIN_DEPLOYMENT", "domain_purchase_required_for_validation": False}},
|
|
{"category": "outbound sales", "status": CapabilityStatus.MISSING, "rationale": "No compliant outreach/sequence/customer contact system exists and V0 forbids outreach.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {}},
|
|
{"category": "CRM", "status": CapabilityStatus.MISSING, "rationale": "No customer pipeline/contact tracking exists.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {}},
|
|
{"category": "payments", "status": CapabilityStatus.PARTIAL, "rationale": "Stripe merchant account is available as CONFIGURED_EXTERNAL_PROVIDER, but no real charges or product checkout integration are implemented in this milestone.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {"payment_provider_state": "CONFIGURED_EXTERNAL_PROVIDER", "implementation_state": "IMPLEMENTATION_MISSING", "real_charges_allowed": False}},
|
|
{"category": "invoicing", "status": CapabilityStatus.MISSING, "rationale": "No invoicing workflow exists.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
|
|
{"category": "customer support", "status": CapabilityStatus.MISSING, "rationale": "No support inbox or customer service workflow exists.", "priority": CapabilityPriority.BEFORE_SCALING, "evidence": {}},
|
|
{"category": "company budget management", "status": CapabilityStatus.MISSING, "rationale": "V0 blocks spend but future validation needs tranche/budget controls.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {}},
|
|
{"category": "legal/compliance", "status": CapabilityStatus.MISSING, "rationale": "No contracts, terms, privacy, or compliance review workflow exists.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
|
|
{"category": "COMPANY_STRUCTURAL_HUMAN_DEPENDENCY", "status": CapabilityStatus.MISSING if structural else CapabilityStatus.AVAILABLE, "rationale": "Structural human dependency is inherent to the company model and should route assisted-only if severe; it is not solved by platform infrastructure.", "priority": CapabilityPriority.BEFORE_VALIDATION, "evidence": {"dependency_causes": structural}},
|
|
]
|
|
|
|
def _portfolio_concentration(self, cohort: VentureCohort) -> dict[str, Any]:
|
|
fingerprints = [self.fingerprint_proposal(member.proposal) for member in cohort.members.select_related("proposal")]
|
|
data = {
|
|
"industry_distribution": dict(Counter(fp.industry for fp in fingerprints)),
|
|
"business_model_distribution": dict(Counter(fp.business_model for fp in fingerprints)),
|
|
"primary_channel_distribution": dict(Counter(fp.primary_distribution_channel for fp in fingerprints)),
|
|
"icp_distribution": dict(Counter(fp.icp[:80] for fp in fingerprints)),
|
|
"capability_dependency_distribution": dict(Counter(cap for fp in fingerprints for cap in fp.required_capability_set)),
|
|
}
|
|
flags = []
|
|
for key, counts in data.items():
|
|
if counts and max(counts.values()) >= max(4, int(cohort.cohort_size * 0.6)):
|
|
flags.append({"type": "PORTFOLIO_CONCENTRATION", "dimension": key, "value": max(counts, key=counts.get), "count": max(counts.values())})
|
|
data["flags"] = flags
|
|
return data
|
|
|
|
def _generic_concentration_penalty(self, proposal: CompanyProposal) -> int:
|
|
text = " ".join([proposal.title, proposal.description, proposal.proposed_solution, proposal.business_model, proposal.acquisition_strategy]).lower()
|
|
penalty = 0
|
|
if "shopify" in text or "ecommerce" in text:
|
|
penalty += 3
|
|
if "audit" in text and not any(term in text for term in ["ai", "agent", "platform", "monitor", "automation"]):
|
|
penalty += 4
|
|
if "emergency" in text or "fix" in text:
|
|
penalty += 4
|
|
if "consulting" in text or "one-time" in text:
|
|
penalty += 3
|
|
if "outbound" in text or "community" in text:
|
|
penalty += 2
|
|
return penalty
|
|
|
|
def _collision_summary(self, cohort: VentureCohort) -> dict[str, int]:
|
|
counts = Counter(cohort.collisions.values_list("classification", flat=True))
|
|
return {choice: counts.get(choice, 0) for choice in OverlapClassification.values}
|
|
|
|
def _readable_cohort_report(self, content: dict[str, Any]) -> str:
|
|
ranking = "\n".join(f"{row['rank']}. {row['company']} - score {row['ic_score']}, P($500) {row['probability']}%, {row['decision']}" for row in content["rankings"])
|
|
top = "\n".join(f"- {row['company']}: {row['thesis']}" for row in content["top_3"])
|
|
demand = "\n".join(f"- {row['capability']}: {row['count']} companies, earliest {row['earliest_stage']}" for row in content["capability_demand"][:15])
|
|
rejections = "\n".join(f"- Slot {row['slot_index']} attempt {row['attempt']}: {row['decision']} - {row['candidate_title']} ({row['reason']})" for row in content["generation_rejections"]) or "- none"
|
|
clusters = "\n".join(f"- {row['company']}: {row['portfolio_thesis']} ({row['classification']}, {row['similarity_score']})" for row in content["thesis_clusters"]) or "- none"
|
|
saturation = "\n".join(f"- {row['thesis']}: {row['status_recommendation']} ({row['proposal_count']} proposals, best {row['best_ic_score']})" for row in content["saturation_analysis"]) or "- none"
|
|
return f"# Venture Discovery Cohort Report\n\nCohort: {content['cohort_id']}\nMandate: {content['mandate']}\nTotal spend: $0\nCustomer outreach: none\n\n## Ideation Mandate\n{json.dumps(content['ideation_mandate'], indent=2)}\n\n## Hard Exclusions\n" + "\n".join(f"- {item}" for item in content["hard_exclusions"]) + "\n\n## Soft Exclusions\n" + "\n".join(f"- {item}" for item in content["soft_exclusions"]) + "\n\n## Search Territories\n" + "\n".join(f"- {item}" for item in content["search_territories"]) + f"\n\n## Generation Rejections\n{rejections}\n\n## Thesis Registry Before\n{json.dumps(content['thesis_registry_before'], indent=2)}\n\n## Thesis Registry After\n{json.dumps(content['thesis_registry_after'], indent=2)}\n\n## Thesis Clusters\n{clusters}\n\n## Saturation Analysis\n{saturation}\n\n## Idea Diversity Metrics\n{json.dumps(content['idea_diversity_metrics'], indent=2)}\n\n## Ranking\n{ranking}\n\n## Top 3 Finalists\n{top}\n\n## Collisions\n{json.dumps(content['collisions'], indent=2)}\n\n## Portfolio Concentration\n{json.dumps(content['portfolio_concentration'], indent=2)}\n\n## Capability Demand\n{demand}\n\n## Recommended Artifex Build Priorities\n" + "\n".join(f"- {item}" for item in content["recommended_build_priorities"])
|
|
|
|
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_pitch(self, pitch: dict[str, Any]) -> str:
|
|
return "\n\n".join(f"{section}\n{pitch.get(section, '')}" for section in PITCH_SECTIONS)
|
|
|
|
def _readable_mandate(self, mandate: CompanyMandate) -> str:
|
|
return f"Objective\n{mandate.objective}\n\nConstraints\n{json.dumps(mandate.constraints, indent=2)}\n\nOptimization targets\n" + "\n".join(f"- {item}" for item in mandate.optimization_targets)
|
|
|
|
def _readable_board(self, review: CompanyBoardReview) -> str:
|
|
return f"Recommendation: {review.recommendation}\n\nStrengths\n" + "\n".join(f"- {item}" for item in review.strengths) + "\n\nWeaknesses\n" + "\n".join(f"- {item}" for item in review.weaknesses)
|
|
|
|
def _readable_responses(self, responses: list[ICResponse]) -> str:
|
|
return "\n\n".join(f"Q: {response.question.question}\nA: {response.answer}\nUncertainty: {response.uncertainty}" for response in responses)
|
|
|
|
def _readable_score(self, decision: ICDecision) -> str:
|
|
scores = "\n".join(f"- {key}: {value}/100" for key, value in decision.component_scores.items())
|
|
return f"Decision: {decision.decision}\nComposite: {decision.composite_score}/100\nP($500 within 30 days): {decision.probability_500_within_30_days}%\n\nScores\n{scores}\n\nCondition\n{decision.validation_condition}"
|
|
|
|
def _readable_research(self, research: dict[str, Any]) -> str:
|
|
sources = "\n".join(f"- {source.get('category')}: {source.get('title')} {source.get('url')}" for source in research.get("sources", []))
|
|
missing = ", ".join(research.get("unverified_categories", [])) or "none"
|
|
return f"Coverage\n{json.dumps(research.get('coverage', {}), indent=2)}\n\nSources\n{sources}\n\nUnverified categories\n{missing}"
|
|
|
|
def _readable_capability_gap(self, available: list[str], partial: list[str], missing: list[str], ranked: list[dict[str, str]]) -> str:
|
|
return "AVAILABLE\n" + "\n".join(f"- {item}" for item in available) + "\n\nPARTIAL\n" + "\n".join(f"- {item}" for item in partial) + "\n\nMISSING\n" + "\n".join(f"- {item}" for item in missing) + "\n\nNEXT ARTIFEX CAPABILITIES REQUIRED\n" + "\n".join(f"- {item['category']} ({item['priority']})" for item in ranked)
|
|
|
|
def _readable_memo(self, memo: dict[str, Any]) -> str:
|
|
return json.dumps(memo, indent=2)
|
|
|
|
def _fingerprint(self, payload: dict[str, Any]) -> str:
|
|
return hashlib.sha256((str(payload["title"]).lower() + str(payload["target_customer"]).lower() + str(payload["business_model"]).lower()).encode()).hexdigest()
|