102 lines
6.6 KiB
Python
102 lines
6.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from agents.venture_discovery import EVIDENCE_CEILINGS, SCORE_DEFINITIONS, SCORE_DIMENSIONS, VentureDiscoveryService
|
|
from control_plane.ventures.models import CompanyProposal, EvidenceTier, OverlapClassification, PortfolioICReview, VentureCapabilityDemand, VentureCohort, VentureCollision, VentureThesisFingerprint
|
|
from graph.bootstrap import champion_venture_discovery_cohort_graph_v1
|
|
from graph.langgraph_runtime import LangGraphRuntime
|
|
from graph.models import GraphRun, GraphRunStatus
|
|
from graph.venture_cohort import venture_discovery_cohort_registry
|
|
from model_router.router import ModelProvider, ModelRequestContract, ModelResponseContract, ModelRouter
|
|
|
|
|
|
class SequenceProvider(ModelProvider):
|
|
provider_name = "sol-sequence"
|
|
|
|
def __init__(self) -> None:
|
|
self.company_calls = 0
|
|
self.research_calls = 0
|
|
|
|
def complete(self, request: ModelRequestContract) -> ModelResponseContract:
|
|
if "Bounded public web market research" in request.prompt:
|
|
self.research_calls += 1
|
|
categories = ["competitors", "pricing", "customer_pain", "market_alternatives", "regulatory_platform_risks"]
|
|
return ModelResponseContract("luna", json.dumps({"sources": [{"url": f"https://example.com/{self.research_calls}/{cat}", "title": cat, "category": cat, "summary": cat} for cat in categories], "findings": {cat: [cat] for cat in categories}, "coverage": {cat: True for cat in categories}}), {})
|
|
self.company_calls += 1
|
|
n = self.company_calls
|
|
industries = ["Shopify", "Developer", "Legal", "Healthcare", "Real Estate", "Restaurant", "Security", "Education", "Logistics", "Finance"]
|
|
industry = industries[(n - 1) % len(industries)]
|
|
return ModelResponseContract("sol", json.dumps({"title": f"{industry} Validation Offer {n}", "one_line_thesis": f"Sell a productized {industry.lower()} audit to a narrow buyer before building software.", "description": f"A fixed-scope {industry} audit.", "problem": f"{industry} buyers have urgent operational gaps.", "target_customer": f"Small {industry} operators with active revenue.", "proposed_solution": f"Manual {industry} audit with prioritized fixes.", "business_model": "Productized service", "pricing_hypothesis": "$99 audit", "acquisition_strategy": "Compliant community posts and direct referrals after approval", "validation_plan": "Collect 5 credible target-customer responses or 1 willingness-to-pay signal before build/spend.", "capital_requested": "50", "time_to_first_dollar_estimate": "3-7 days after outreach approval", "expected_margin": "80-90% gross margin", "build_complexity": "LOW", "market_evidence": [], "differentiation": "Fast narrow audit", "major_risks": ["Demand unproven"], "confidence": 0.65}), {})
|
|
|
|
def health(self) -> str:
|
|
return "AVAILABLE"
|
|
|
|
|
|
def service() -> VentureDiscoveryService:
|
|
provider = SequenceProvider()
|
|
return VentureDiscoveryService(ModelRouter({"sol": provider, "luna": provider}), web_research_available=True)
|
|
|
|
|
|
def test_evidence_tiers_probability_ceiling_and_score_orientation() -> None:
|
|
svc = service()
|
|
proposal = svc.generate_single_company(svc.create_v0_mandate())
|
|
assert proposal.evidence_tier == EvidenceTier.TIER_0_THESIS
|
|
assert svc.calibrate_probability(proposal, raw_probability=80)["evidence_adjusted_probability"] == EVIDENCE_CEILINGS[EvidenceTier.TIER_0_THESIS]
|
|
svc.conduct_market_research(proposal)
|
|
assert proposal.evidence_tier == EvidenceTier.TIER_1_PUBLIC_EVIDENCE
|
|
assert svc.calibrate_probability(proposal, raw_probability=80)["evidence_adjusted_probability"] == EVIDENCE_CEILINGS[EvidenceTier.TIER_1_PUBLIC_EVIDENCE]
|
|
diligence = svc.start_ic_diligence(proposal)
|
|
svc.generate_ic_questions(diligence)
|
|
svc.answer_questions(diligence)
|
|
svc.red_team(diligence)
|
|
svc.final_company_response(diligence)
|
|
decision = svc.score_and_decide(diligence)
|
|
assert set(decision.component_scores) == set(SCORE_DIMENSIONS)
|
|
assert all("100 = highly attractive" in definition for definition in decision.score_definitions.values())
|
|
assert decision.probability_500_within_30_days <= decision.evidence_ceiling
|
|
assert decision.raw_probability_500_within_30_days >= decision.probability_500_within_30_days
|
|
|
|
|
|
def test_identity_contamination_detection_and_fingerprint_collision() -> None:
|
|
svc = service()
|
|
proposal = svc.generate_single_company(svc.create_v0_mandate())
|
|
contaminated = {"Company": "Unrelated CRM SaaS", "ICP": "enterprise banks", "Problem": "loan defaults", "Product": "risk platform"}
|
|
result = svc.validate_identity_content(proposal, contaminated)
|
|
assert result["passed"] is False
|
|
fp = svc.fingerprint_proposal(proposal)
|
|
assert isinstance(fp, VentureThesisFingerprint)
|
|
other = svc.generate_single_company(proposal.mandate)
|
|
other.title = proposal.title + " Copy"
|
|
other.target_customer = proposal.target_customer
|
|
other.problem = proposal.problem
|
|
other.proposed_solution = proposal.proposed_solution
|
|
other.business_model = proposal.business_model
|
|
other.save(update_fields=["title", "target_customer", "problem", "proposed_solution", "business_model", "updated_at"])
|
|
overlap = svc.classify_overlap(proposal, other)
|
|
assert overlap["classification"] in {OverlapClassification.NEAR_DUPLICATE, OverlapClassification.DUPLICATE, OverlapClassification.COMPETITIVE}
|
|
|
|
|
|
def test_cohort_size_ranking_top3_capability_aggregation_and_graph_lineage() -> None:
|
|
svc = service()
|
|
version = champion_venture_discovery_cohort_graph_v1()
|
|
graph_run = GraphRun.objects.create(execution_graph_version=version, current_node=version.graph_spec["entry"])
|
|
LangGraphRuntime(venture_discovery_cohort_registry(svc, cohort_size=10, concurrency=2)).run_until_terminal_or_paused(graph_run)
|
|
graph_run.refresh_from_db()
|
|
cohort = VentureCohort.objects.get(id=graph_run.metadata["cohort_id"])
|
|
|
|
assert graph_run.status == GraphRunStatus.COMPLETE
|
|
assert cohort.members.count() == 10
|
|
assert CompanyProposal.objects.count() == 10
|
|
assert cohort.members.filter(is_top_3=True).count() == 3
|
|
assert PortfolioICReview.objects.filter(cohort=cohort).exists()
|
|
assert VentureCollision.objects.filter(cohort=cohort).count() == 45
|
|
assert VentureCapabilityDemand.objects.filter(cohort=cohort).exists()
|
|
assert cohort.metadata["real_spend"] == 0
|
|
assert cohort.metadata["real_customer_outreach"] is False
|
|
assert cohort.concurrency == 2
|
|
assert cohort.metrics["peak_concurrency"] == 2
|
|
report = cohort.mandate.artifacts.get(artifact_type="VENTURE_DISCOVERY_COHORT_REPORT")
|
|
assert len(report.content["rankings"]) == 10
|
|
assert len(report.content["top_3"]) == 3
|
|
assert report.content["total_spend"] == 0
|