1134 lines
110 KiB
Python
1134 lines
110 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 CapabilityPriority, CapabilityStatus, CohortIdeationMandate, CompanyBoardReview, CompanyCapabilityRequirement, CompanyMandate, CompanyProposal, CompanyProposalStatus, EvidenceTier, ICDecision, ICDecisionType, ICDiligence, ICQuestion, ICResponse, NoveltyGateDecision, OpportunityTerritory, OverlapClassification, PortfolioCapabilityGap, PortfolioICReview, PortfolioSaturationAnalysis, PortfolioThesis, PortfolioThesisCluster, PortfolioThesisStatus, ThesisMatchClassification, VentureArtifact, VentureCapabilityDemand, VentureCohort, VentureCohortMember, VentureCollision, VentureGenerationRejection, VentureThesis, VentureThesisFingerprint
|
|
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", "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}
|
|
|
|
|
|
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 could plausibly turn at most $50 of external validation capital into $500 net new cash.",
|
|
constraints={"external_validation_capital_max": 50, "target_net_new_cash": 500, "target_window_days": 30, "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},
|
|
optimization_targets=["time to first dollar", "capital efficiency", "real demand evidence", "high gross margin", "realistic execution", "low external dependency", "ability to validate cheaply", "AI leverage where economically justified", "service-to-platform evolution potential"],
|
|
metadata={"milestone": "VENTURE_DISCOVERY_V0", "spend_authorized": False, "customer_outreach_authorized": False, "ai_native_policy": AI_NATIVE_POLICY},
|
|
)
|
|
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},
|
|
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) -> dict[str, Any]:
|
|
required = ["competitors", "pricing", "customer_pain", "market_alternatives", "regulatory_platform_risks"]
|
|
research = {"coverage": {category: False for category in required}, "sources": [], "findings": {}, "unverified_categories": required, "research_available": False}
|
|
search_sources = self._searxng_sources(proposal, required) if self.web_research_available else []
|
|
page_corpus = self._page_corpus(search_sources) if search_sources else []
|
|
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 source, mark coverage false. 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"))
|
|
source_categories = {str(source.get("category", "")) for source in research.get("sources", []) if source.get("url")}
|
|
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)
|
|
proposal.market_evidence = [*self._as_list(proposal.market_evidence), *research.get("sources", []), {"type": "research_coverage", "source": "venture_research", "summary": f"Source-linked research coverage: {round(coverage_ratio * 100)}%", "coverage": coverage, "unverified_categories": research["unverified_categories"]}]
|
|
proposal.metadata = {**proposal.metadata, "research": {"coverage_ratio": coverage_ratio, "unverified_categories": research["unverified_categories"], "source_count": len(research.get("sources", [])), "page_fetch_count": research.get("page_fetch_count", 0), "provider": research.get("provider", "none"), "search_provider": research.get("search_provider", "none")}}
|
|
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", "Bounded Public Market Research", 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:
|
|
scores = self._evidence_scores(diligence.proposal)
|
|
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.create(diligence=diligence, 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"VDV03-{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 v2", "company": "venture_discovery v1"}, research_policy={"stage_a": "lightweight_all", "deeper_research": "top_5_if_required", "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.3"})
|
|
|
|
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)
|
|
max_attempts = int(cohort.scoring_policy.get("duplicate_policy", DEFAULT_DUPLICATE_POLICY).get("max_attempts_per_slot", 3))
|
|
attempts = hard_rejections = duplicate_rejections = soft_reviews = failed_slots = model_requests = 0
|
|
for index in range(cohort.cohort_size):
|
|
territory = mandate.opportunity_territories[index % len(mandate.opportunity_territories)] if mandate.opportunity_territories else OpportunityTerritory.OPEN_CATEGORY
|
|
accepted: tuple[dict[str, Any], str, dict[str, Any]] | None = None
|
|
for attempt in range(1, max_attempts + 1):
|
|
attempts += 1
|
|
model_requests += 1
|
|
context = {"hard_exclusions": mandate.hard_exclusions, "soft_exclusions": mandate.soft_exclusions, "territory": territory, "brief": self.ideation_mandate_payload(mandate)}
|
|
payload, source = self._company_payload(cohort.mandate, ideation_index=index + 1, ideation_context=context)
|
|
gate = self.novelty_gate(payload, cohort, accepted_payloads, mandate, territory=str(territory))
|
|
if gate["decision"] == NoveltyGateDecision.ACCEPT:
|
|
accepted = (payload, source, gate)
|
|
break
|
|
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=index + 1, attempt=attempt, decision=gate["decision"], reason=gate["explanation"], candidate=payload, matched_thesis=gate.get("matched_thesis"), similarity_score=float(gate.get("similarity_score", 0.0)), metadata={"territory": str(territory), "classification": gate.get("classification")})
|
|
if accepted is None:
|
|
failed_slots += 1
|
|
VentureGenerationRejection.objects.create(cohort=cohort, slot_index=index + 1, attempt=max_attempts, decision=NoveltyGateDecision.FAILED_IDEATION, reason="max ideation attempts exhausted", metadata={"territory": str(territory)})
|
|
continue
|
|
payload, source, gate = accepted
|
|
proposal = self.generate_single_company(cohort.mandate, ideation_index=index + 1, payload=payload, source=source)
|
|
generation_index = index + 1
|
|
proposal.metadata = {**proposal.metadata, "cohort_id": cohort.cohort_id, "independent_generation_index": generation_index, "prior_ideas_visible": False, "search_territory": str(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})
|
|
proposals.append(proposal)
|
|
accepted_payloads.append(payload)
|
|
cohort.status = "PROPOSALS_GENERATED"
|
|
cohort.metrics = {**cohort.metrics, "requested_company_count": cohort.cohort_size, "generation_attempts": attempts, "accepted_proposals": len(proposals), "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": 1, "peak_concurrency": max(cohort.metrics.get("peak_concurrency", 0), 1)}
|
|
cohort.save(update_fields=["metrics", "status", "updated_at"])
|
|
if len(proposals) != cohort.cohort_size:
|
|
cohort.status = "FAILED_IDEATION"
|
|
cohort.save(update_fields=["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).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
|
|
cohort.metrics = {**cohort.metrics, "research_runtime_seconds": round(time.monotonic() - started, 2), "total_sources": sum(source_counts), "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)
|
|
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 portfolio_ic(self, cohort: VentureCohort) -> PortfolioICReview:
|
|
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
|
|
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)
|
|
score = round(decision.composite_score + decision.probability_500_within_30_days * 0.2 + ai_bonus - capability_burden * 1.5 - collision_risk[str(proposal.id)] * 2 - concentration_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, "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)
|
|
for rank, row in enumerate(rows, start=1):
|
|
member = cohort.members.get(proposal_id=row["proposal_id"])
|
|
member.rank = rank
|
|
member.is_top_3 = rank <= 3
|
|
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": rows[:3], "concentration": concentration, "metadata": {"no_funding": True}})
|
|
cohort.status = "PORTFOLIO_IC_COMPLETE"
|
|
cohort.save(update_fields=["status", "updated_at"])
|
|
return review
|
|
|
|
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
|
|
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", "metadata", "updated_at"])
|
|
updates.append({"thesis": thesis.canonical_name, "status": thesis.status, "proposal_count": thesis.proposal_count, "best_ic_score": thesis.best_ic_score})
|
|
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")]
|
|
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")]
|
|
content = {"cohort_id": cohort.cohort_id, "mandate": cohort.mandate.objective, "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"
|
|
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 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.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)
|
|
best_score = 0.0
|
|
best_name = ""
|
|
for accepted in accepted_payloads:
|
|
accepted_name = self.canonical_thesis_name(accepted)
|
|
if accepted_name == canonical:
|
|
return {"matched_thesis": None, "similarity_score": 1.0, "classification": ThesisMatchClassification.DUPLICATE, "explanation": f"Current cohort already accepted canonical thesis {accepted_name}."}
|
|
score = 1.0 if accepted_name == canonical 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}."}
|
|
|
|
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"
|
|
title = str(payload.get("title", "Untitled thesis")).strip()
|
|
return re.sub(r"\s+", " ", title)[:255]
|
|
|
|
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 Venture Discovery V0.3. Return a single JSON object, not a list. 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. Prefer AI wrapper platforms, agentic workflow products, vertical copilots, AI-enabled services with low human labor, developer/AI infrastructure, intelligence products, automation products, service-to-platform paths, recurring revenue, and owned-compute leverage. 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")) + ". This is a creative search constraint, not a fixed solution. 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, 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"]
|
|
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]) -> list[dict[str, Any]]:
|
|
client = self.search_client or SearxngSearchClient.from_resources()
|
|
if client is None:
|
|
return []
|
|
sources = []
|
|
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.target_customer} {query_terms.get(category, category)}"
|
|
try:
|
|
sources.extend(client.search(query, category=category, limit=3))
|
|
except Exception:
|
|
continue
|
|
return sources[:15]
|
|
|
|
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 _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)
|
|
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, "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
|
|
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": "Deployment planning exists, but company-specific production deployment workflow is not implemented.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
|
|
{"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.MISSING, "rationale": "No payment collection system exists.", "priority": CapabilityPriority.BEFORE_FIRST_CUSTOMER, "evidence": {}},
|
|
{"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": {}},
|
|
]
|
|
|
|
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()
|