761 lines
74 KiB
Python
761 lines
74 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.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, CompanyBoardReview, CompanyCapabilityRequirement, CompanyMandate, CompanyProposal, CompanyProposalStatus, EvidenceTier, ICDecision, ICDecisionType, ICDiligence, ICQuestion, ICResponse, OverlapClassification, PortfolioCapabilityGap, PortfolioICReview, VentureArtifact, VentureCapabilityDemand, VentureCohort, VentureCohortMember, VentureCollision, 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."}
|
|
|
|
|
|
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) -> CompanyProposal:
|
|
payload, source = self._company_payload(mandate, ideation_index=ideation_index)
|
|
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:
|
|
mandate = self.create_v0_mandate()
|
|
cohort_id = f"VDV02-{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 v1", "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}, evidence_calibration_policy={tier: ceiling for tier, ceiling in EVIDENCE_CEILINGS.items()}, metadata={"real_spend": 0, "real_customer_outreach": False})
|
|
|
|
def generate_independent_proposals(self, cohort: VentureCohort) -> list[CompanyProposal]:
|
|
started = time.monotonic()
|
|
|
|
def generate(index: int) -> tuple[int, str]:
|
|
proposal = self.generate_single_company(cohort.mandate, ideation_index=index + 1)
|
|
return index, str(proposal.id)
|
|
|
|
generated = self._bounded_map(range(cohort.cohort_size), generate, cohort.concurrency)
|
|
peak_concurrency = self._last_bounded_map_peak
|
|
proposals = []
|
|
for index, proposal_id in sorted(generated, key=lambda item: item[0]):
|
|
proposal = CompanyProposal.objects.get(id=proposal_id)
|
|
generation_index = index + 1
|
|
proposal.metadata = {**proposal.metadata, "cohort_id": cohort.cohort_id, "independent_generation_index": generation_index, "prior_ideas_visible": False}
|
|
proposal.save(update_fields=["metadata", "updated_at"])
|
|
VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"generation_index": generation_index})
|
|
proposals.append(proposal)
|
|
cohort.status = "PROPOSALS_GENERATED"
|
|
cohort.metrics = {**cohort.metrics, "proposal_generation_runtime_seconds": round(time.monotonic() - started, 2), "proposal_generation_peak_concurrency": peak_concurrency}
|
|
cohort.save(update_fields=["metrics", "status", "updated_at"])
|
|
return proposals
|
|
|
|
def research_cohort(self, cohort: VentureCohort) -> None:
|
|
started = time.monotonic()
|
|
|
|
def research(member_id: str) -> int:
|
|
member = VentureCohortMember.objects.select_related("proposal").get(id=member_id)
|
|
return len(self.conduct_market_research(member.proposal).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 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 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
|
|
content = {"cohort_id": cohort.cohort_id, "mandate": cohort.mandate.objective, "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:
|
|
existing = getattr(proposal, "fingerprint", 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 _company_payload(self, mandate: CompanyMandate, *, ideation_index: int | 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 ""
|
|
response = self.router.complete(ModelRequestContract(purpose=ModelCapability.PLANNING, model_hint=self.ideation_model_hint, prompt="Generate exactly ONE startup idea for Venture Discovery V0. 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, but do not require, AI-native businesses where Artifex can deliver most value using agents/local inference: AI wrapper platforms, AI-enabled services, AI infrastructure/developer tools, AI intelligence products, or SMB/enterprise workflow automation. Do not let AI novelty outweigh customer demand. Penalize generic audits, emergency fix services, one-off consulting, Shopify/ecommerce concentration, and identical outbound-led service models unless exceptional. 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." + slot + " Mandate: " + json.dumps({"objective": mandate.objective, "constraints": mandate.constraints, "optimization_targets": mandate.optimization_targets, "ai_native_policy": AI_NATIVE_POLICY})))
|
|
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"))
|
|
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 Sol or public web research." if fallback else "Generated by Sol; public web research still 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])
|
|
return f"# Venture Discovery Cohort Report\n\nCohort: {content['cohort_id']}\nMandate: {content['mandate']}\nTotal spend: $0\nCustomer outreach: none\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()
|