Janady13/AI_Brain / services/seeder_api/intelligence.py

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/AI_Brain / services/seeder_api/intelligence.py
Date	Not specified in source
Repository	Janady13/AI_Brain
Revision	50285aff347685f043e7a72e4d72891eebaa6cd6
Branch	codex/build-online-github-ani-hrtsoc
Path or archive member	services/seeder_api/intelligence.py
Original SHA-256	d2fdddb7dbba561e0e1271d0c7b00888c7fb98f463ad063073a1ce38bcaee85a
Git blob SHA	c4d049273a0b00919ddc332ce0537f6e0ac0e554
Reading-copy SHA-256	5c2240ba5cdc117f3c3efa405b4b0977fd978d82c291ebf2a8c4401a63ba7ea7
Original source ↗
Every recorded occurrence

These 3 records contain identical original source bytes. Their paths and revisions remain part of the archive.
Original record	Branch or archive	Revision / identifier	Original location
Janady13/AI_Brain / services/seeder_api/intelligence.py	codex/build-online-github-ani-hrtsoc	50285aff347685f043e7a72e4d72891eebaa6cd6	Open original
Janady13/AI_Brain / services/seeder_api/intelligence.py	copilot/vscode1760283895982	9e1c1a5b171addb5bb4a83b2ea7dede50c1cc81e	Open original
Janady13/AI_Brain / services/seeder_api/intelligence.py	codex/build-online-github-ani-hrtsoc	50285aff347685f043e7a72e4d72891eebaa6cd6	Open original
Recovered source transcript

Download complete source text ↓10,463 lines · 396.6 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
"""Marketing intelligence engine that ranks and scores enumerated campaigns."""

from __future__ import annotations

import base64
import importlib
import json
import math
import os
import re
from dataclasses import dataclass
from functools import lru_cache
from statistics import mean
from typing import Any, Dict, List, Sequence, Tuple

try:  # pragma: no cover - optional dependency that may not be installed in tests
    from google.ads.googleads.client import GoogleAdsClient  # type: ignore
except ImportError:  # pragma: no cover - gracefully handled when SDK missing
    GoogleAdsClient = None  # type: ignore

try:  # pragma: no cover - optional dependency that may not be installed in tests
    from google.oauth2 import service_account  # type: ignore
except ImportError:  # pragma: no cover - gracefully handled when SDK missing
    service_account = None  # type: ignore

try:  # pragma: no cover - optional dependency that may not be installed in tests
    import stripe  # type: ignore
except ImportError:  # pragma: no cover - gracefully handled when SDK missing
    stripe = None  # type: ignore

try:  # pragma: no cover - optional dependency that may not be installed in tests
    from paypalcheckoutsdk.core import (  # type: ignore
        PayPalHttpClient,
        LiveEnvironment,
        SandboxEnvironment,
    )
except ImportError:  # pragma: no cover - gracefully handled when SDK missing
    PayPalHttpClient = None  # type: ignore
    LiveEnvironment = None  # type: ignore
    SandboxEnvironment = None  # type: ignore

try:  # pragma: no cover - optional dependency for AWS integrations
    import boto3  # type: ignore
except ImportError:  # pragma: no cover - gracefully handled when SDK missing
    boto3 = None  # type: ignore

try:  # pragma: no cover - optional dependency for Postgres persistence
    import psycopg2  # type: ignore
    from psycopg2.extras import Json  # type: ignore
except Exception:  # pragma: no cover - handle environments without psycopg2
    psycopg2 = None  # type: ignore
    Json = None  # type: ignore


MARKETING_ENUMERATION = os.path.join(
    os.path.dirname(__file__), "..", "..", "daemon", "enumerations", "marketing_500.txt"
)
MARKETING_BENCHMARKS = os.path.join(
    os.path.dirname(__file__), "..", "..", "daemon", "datasets", "marketing_benchmarks.json"
)
CJ_OFFERS_DATASET = os.path.join(
    os.path.dirname(__file__), "..", "..", "daemon", "datasets", "cj_offers.json"
)
AMAZON_MARKETPLACE_DATASET = os.path.join(
    os.path.dirname(__file__), "..", "..", "daemon", "datasets", "amazon_marketplace.json"
)
EBAY_MARKETPLACE_DATASET = os.path.join(
    os.path.dirname(__file__), "..", "..", "daemon", "datasets", "ebay_marketplace.json"
)
AWS_GROWTH_PLAYBOOKS = os.path.join(
    os.path.dirname(__file__), "..", "..", "daemon", "datasets", "aws_growth_playbooks.json"
)
CHANNEL_INVESTMENTS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "channel_investments.json",
)

AUTOMATION_SPRINTS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "automation_sprints.json",
)

PERFORMANCE_MEMORY_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "performance_memory.json",
)

COMPETITIVE_LANDSCAPE_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "competitive_landscape.json",
)

INFLUENCER_PARTNERS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "influencer_partners.json",
)

STOCK_MARKETS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "global_stock_markets.json",
)

STATISTICAL_PRIORS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "statistical_reasoning.json",
)

COGNITIVE_CAPABILITIES_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "cognitive_capabilities.json",
)

AGENTIC_SYSTEMS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "agentic_systems.json",
)

KNOWLEDGE_WORKFLOWS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "knowledge_workflows.json",
)

LANGUAGE_CAPABILITIES_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "language_capabilities.json",
)

PAYMENT_PLATFORMS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "payment_platforms.json",
)

PHYSICS_ECONOMICS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "physics_economics.json",
)

CODE_AUTOMATION_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "code_automation.json",
)

MOE_EXPERTS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "moe_experts.json",
)

AGENTIC_LOOPS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "agentic_loops.json",
)

QUANTUM_SIMULATIONS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "quantum_simulations.json",
)

INVESTING_PLATFORMS_DATASET = os.path.join(
    os.path.dirname(__file__),
    "..",
    "..",
    "daemon",
    "datasets",
    "investing_platforms.json",
AWS_GROWTH_PLAYBOOKS = os.path.join(
    os.path.dirname(__file__), "..", "..", "daemon", "datasets", "aws_growth_playbooks.json"
)

requests_spec = importlib.util.find_spec("requests")
requests = importlib.import_module("requests") if requests_spec else None


def _load_lines(path: str) -> List[str]:
    with open(path, "r", encoding="utf-8") as handle:
        return [line.strip() for line in handle if line.strip()]


@dataclass(slots=True)
class MarketingCampaign:
    """Structured representation for one marketing campaign idea."""

    raw: str
    channel: str
    audience: str
    offer: str
    objective: str

    @classmethod
    def parse(cls, raw: str) -> "MarketingCampaign":
        text = raw.strip().rstrip(".")
        if not text.lower().startswith("campaign:"):
            raise ValueError(f"Unsupported campaign format: {raw}")

        body = text[len("Campaign:") :].strip()

        try:
            channel, remainder = body.split("targeting", 1)
            audience, remainder = remainder.split("leveraging", 1)
            offer, objective = remainder.split(" to ", 1)
        except ValueError as exc:  # pragma: no cover - enumeration should be well formed
            raise ValueError(f"Malformed campaign sentence: {raw}") from exc

        return cls(
            raw=raw,
            channel=channel.strip(),
            audience=audience.strip(),
            offer=offer.strip(),
            objective=objective.strip().rstrip("."),
        )


@dataclass(slots=True)
class CampaignMetrics:
    """Statistical expectations for a marketing campaign."""

    expected_roi: float
    expected_ctr: float
    confidence: float
    drivers: Dict[str, float]


@dataclass(slots=True)
class StatisticalPrior:
    """Historical statistical benchmark for a marketing dimension."""

    feature: str
    feature_type: str
    mean_roi: float
    std_roi: float
    mean_ctr: float
    std_ctr: float
    sample_size: int
    weight: float
    baseline_budget: float
    notes: str


@dataclass(slots=True)
class CognitiveCapability:
    """Represents a cognitive pillar the ANI can operationalize."""

    name: str
    pillar: str
    description: str
    signal_keywords: List[str]
    prompt_strategies: List[str]
    frameworks: List[str]
    tooling: List[str]
    metrics: List[str]
    knowledge_assets: List[str]
    play: str

    @classmethod
    def from_dict(cls, payload: Dict[str, Any]) -> "CognitiveCapability":
        return cls(
            name=payload["name"],
            pillar=payload["pillar"],
            description=payload["description"],
            signal_keywords=[keyword.lower() for keyword in payload["signal_keywords"]],
            prompt_strategies=payload["prompt_strategies"],
            frameworks=payload["frameworks"],
            tooling=payload["tooling"],
            metrics=payload["metrics"],
            knowledge_assets=payload["knowledge_assets"],
            play=payload["play"],
        )


@dataclass(slots=True)
class CognitiveCapabilityPlan:
    """Cognitive orchestration outcome derived from prompt and campaign context."""

    capability: str
    pillar: str
    maturity: str
    description: str
    reasoning: str
    prompt_strategy: str
    frameworks: List[str]
    tooling: List[str]
    metrics: List[str]
    knowledge_assets: List[str]
    orchestration_steps: List[str]
class BudgetProfile:
    """Baseline investment expectations for a marketing channel."""

    channel: str
    keywords: List[str]
    baseline_cpc: float
    baseline_cpl: float
    avg_conversion_rate: float
    avg_roi: float
    recommended_monthly_spend: float


@dataclass(slots=True)
class BudgetAllocation:
    """Recommended allocation for a marketing campaign channel."""

    channel: str
    audience_focus: str
    offer: str
    objective: str
    allocation_usd: float
    allocation_weight: float
    expected_leads: float
    expected_revenue: float
    rationale: str
    source: str
    notes: str | None = None


@dataclass(slots=True)
class BudgetPlan:
    """Portfolio-level budget recommendation summary."""

    total_budget: float
    allocations: List[BudgetAllocation]
    expected_revenue: float
    expected_roi: float
    payback_period_weeks: float
    summary: str


@dataclass(slots=True)
class AutomationTask:
    """Time-boxed task within an automation sprint."""

    start_minute: int
    end_minute: int
    activity: str
    outcome: str
    tooling: List[str]
    kpis: List[str]
    automation_level: str
    source: str


@dataclass(slots=True)
class AgenticCapability:
    """Represents an agentic system capability that can be activated."""

    name: str
    domain: str
    description: str
    controller: str
    trigger_keywords: List[str]
    frameworks: List[str]
    toolchain: List[str]
    data_interfaces: List[str]
    knowledge_sources: List[str]
    revenue_programs: List[str]
    risk_controls: List[str]
    maturity_signals: List[str]
    base_maturity: str
    metrics: List[str]

    @classmethod
    def from_dict(cls, payload: Dict[str, Any]) -> "AgenticCapability":
        return cls(
            name=payload["name"],
            domain=payload["domain"],
            description=payload["description"],
            controller=payload.get("controller", "Agent controller"),
            trigger_keywords=[keyword.lower() for keyword in payload.get("trigger_keywords", [])],
            frameworks=payload.get("frameworks", []),
            toolchain=payload.get("toolchain", []),
            data_interfaces=payload.get("data_interfaces", []),
            knowledge_sources=payload.get("knowledge_sources", []),
            revenue_programs=payload.get("revenue_programs", []),
            risk_controls=payload.get("risk_controls", []),
            maturity_signals=payload.get("maturity_signals", []),
            base_maturity=payload.get("base_maturity", "foundational"),
            metrics=payload.get("metrics", []),
        )


@dataclass(slots=True)
class AgenticSystemPlan:
    """Structured plan describing how the ANI activates agentic systems."""

    system_name: str
    domain: str
    controller: str
    maturity: str
    reasoning: str
    orchestration_steps: List[str]
    toolchain: List[str]
    data_interfaces: List[str]
    knowledge_sources: List[str]
    revenue_programs: List[str]
    risk_controls: List[str]
    metrics: List[str]
    source: str
class AutomationSprint:
    """Three-hour automation run composed of sequential tasks."""

    total_minutes: int
    stage: str
    focus: str
    summary: str
    tasks: List[AutomationTask]
    notes: str | None = None
    source: str = "automation_playbooks"


@dataclass(slots=True)
class PerformanceMemoryEntry:
    """Historical marketing learning captured from closed-loop experiments."""

    play_name: str
    stage: str
    audience_focus: str
    channel: str
    tags: List[str]
    summary: str
    wins: List[str]
    losses: List[str]
    next_actions: List[str]
    kpi_focus: List[str]
    confidence: float
    source: str = "performance_memory"
    notes: str | None = None


@dataclass(slots=True)
class LanguageCapability:
    """Represents a language-centric capability the ANI can operationalize."""

    name: str
    category: str
    description: str
    signal_keywords: List[str]
    pipeline_components: List[str]
    tooling: List[str]
    datasets: List[str]
    evaluation_metrics: List[str]
    revenue_programs: List[str]
    orchestration_play: str

    @classmethod
    def from_dict(cls, payload: Dict[str, Any]) -> "LanguageCapability":
        return cls(
            name=payload["name"],
            category=payload["category"],
            description=payload["description"],
            signal_keywords=[keyword.lower() for keyword in payload.get("signal_keywords", [])],
            pipeline_components=payload.get("pipeline_components", []),
            tooling=payload.get("tooling", []),
            datasets=payload.get("datasets", []),
            evaluation_metrics=payload.get("evaluation_metrics", []),
            revenue_programs=payload.get("revenue_programs", []),
            orchestration_play=payload.get("orchestration_play", "Operationalize language workflow."),
        )


@dataclass(slots=True)
class LanguageCapabilityPlan:
    """Activation plan for a natural language capability."""

    capability: str
    category: str
    maturity: str
    description: str
    reasoning: str
    pipeline_components: List[str]
    tooling: List[str]
    datasets: List[str]
    evaluation_metrics: List[str]
    revenue_programs: List[str]
    orchestration_steps: List[str]
    source: str
class LearningLoopRecommendation:
    """Actionable recommendation generated from the learning memory."""

    play_name: str
    stage: str
    audience_focus: str
    channel: str
    learning: str
    wins: List[str]
    losses: List[str]
    next_actions: List[str]
    kpi_focus: List[str]
    confidence: float
    source: str
    notes: str | None = None


@dataclass(slots=True)
class ExcellenceSummary:
    """Composite readiness score describing how robust the ANI recommendations are."""

    excellence_index: float
    maturity_level: str
    headline: str
    strengths: List[str]
    watchouts: List[str]
    process_coverage: Dict[str, float]
    notes: str | None = None


@dataclass(slots=True)
class CodeAutomationCapability:
    """Represents AI-driven code and automation capabilities the ANI can deploy."""

    name: str
    category: str
    description: str
    signal_keywords: List[str]
    automation_stages: List[str]
    tooling: List[str]
    platforms: List[str]
    metrics: List[str]
    revenue_programs: List[str]
    guardrails: List[str]
    orchestration_play: str
    base_maturity: str

    @classmethod
    def from_dict(cls, payload: Dict[str, Any]) -> "CodeAutomationCapability":
        return cls(
            name=payload["name"],
            category=payload["category"],
            description=payload["description"],
            signal_keywords=[keyword.lower() for keyword in payload.get("signal_keywords", [])],
            automation_stages=payload.get("automation_stages", []),
            tooling=payload.get("tooling", []),
            platforms=payload.get("platforms", []),
            metrics=payload.get("metrics", []),
            revenue_programs=payload.get("revenue_programs", []),
            guardrails=payload.get("guardrails", []),
            orchestration_play=payload.get(
                "orchestration_play", "Operationalize AI-powered automation workflows."
            ),
            base_maturity=payload.get("base_maturity", "foundational"),
        )


@dataclass(slots=True)
class CodeAutomationPlan:
    """Plan describing how the ANI operationalizes code automation workflows."""

    capability: str
    category: str
    maturity: str
    description: str
    reasoning: str
    automation_stages: List[str]
    tooling: List[str]
    platforms: List[str]
    metrics: List[str]
    revenue_programs: List[str]
    guardrails: List[str]
    orchestration_steps: List[str]
    source: str
class StrategicDirective:
    """Directive that fuses multiple process signals into a prioritized motion."""

    name: str
    priority: int
    objective: str
    primary_channel: str
    supporting_channels: List[str]
    investment_usd: float | None
    expected_revenue_usd: float | None
    confidence: float
    time_horizon_hours: int
    north_star_metric: str
    supporting_metrics: List[str]
    action_steps: List[str]
    dependencies: List[str]
    risk_mitigations: List[str]
    inspiration: str
    source: str = "ani_orchestration"
    notes: str | None = None


@dataclass(slots=True)
class ExecutivePlaybook:
    """Cross-process action plan that guides autonomous execution."""

    headline: str
    focus: str
    urgency: str
    north_star_metric: str
    total_budget_usd: float | None
    expected_revenue_usd: float | None
    confidence: float
    directives: List[StrategicDirective]
    summary: str
    notes: str | None = None


@dataclass(slots=True)
class PhysicsEconomicsCapability:
    """Represents physics, quantum, and economics readiness patterns."""

    name: str
    domain: str
    description: str
    trigger_keywords: List[str]
    marketing_algorithms: List[str]
    physics_principles: List[str]
    quantum_strategies: List[str]
    economic_models: List[str]
    hardware_strategies: List[str]
    data_compression: List[str]
    risk_controls: List[str]
    metrics: List[str]
    maturity_signals: List[str]
    base_maturity: str
    orchestration_play: str
    docs: List[str]

    @classmethod
    def from_dict(cls, payload: Dict[str, Any]) -> "PhysicsEconomicsCapability":
        return cls(
            name=payload["name"],
            domain=payload.get("domain", "physics_economics"),
            description=payload.get("description", "Physics-economics capability."),
            trigger_keywords=[keyword.lower() for keyword in payload.get("trigger_keywords", [])],
            marketing_algorithms=payload.get("marketing_algorithms", []),
            physics_principles=payload.get("physics_principles", []),
            quantum_strategies=payload.get("quantum_strategies", []),
            economic_models=payload.get("economic_models", []),
            hardware_strategies=payload.get("hardware_strategies", []),
            data_compression=payload.get("data_compression", []),
            risk_controls=payload.get("risk_controls", []),
            metrics=payload.get("metrics", []),
            maturity_signals=payload.get("maturity_signals", []),
            base_maturity=payload.get("base_maturity", "emerging"),
            orchestration_play=payload.get(
                "orchestration_play",
                "Operationalize physics and economics telemetry for marketing decisions.",
            ),
            docs=payload.get("docs", []),
        )


@dataclass(slots=True)
class PhysicsEconomicsPlan:
    """Plan describing how the ANI fuses physics and economics intelligence."""

    capability: str
    domain: str
    maturity: str
    description: str
    rationale: str
    marketing_algorithms: List[str]
    physics_principles: List[str]
    quantum_strategies: List[str]
    economic_models: List[str]
    hardware_strategies: List[str]
    data_compression: List[str]
    risk_controls: List[str]
    metrics: List[str]
    orchestration_steps: List[str]
    docs: List[str]
    source: str
class ScenarioProjection:
    """Scenario-level forecast for autonomous marketing execution."""

    scenario: str
    expected_revenue: float
    expected_roi: float
    probability: float
    confidence: float
    accelerators: List[str]
    stabilizers: List[str]
    watchouts: List[str]
    source: str = "scenario_planner"
    notes: str | None = None


@dataclass(slots=True)
class PaymentPlatformCapability:
    """Represents a payment platform knowledge bundle the ANI can deploy."""

    provider: str
    description: str
    integration_focus: List[str]
    docs: List[str]
    sdks: List[str]
    setup: List[str]
    troubleshooting: List[str]
    codefix: List[str]
    marketing_plays: List[str]
    keywords: List[str]
    base_maturity: str
    source: str = "payment_platforms_dataset"

    @classmethod
    def from_dict(cls, payload: Dict[str, Any]) -> "PaymentPlatformCapability":
        return cls(
            provider=payload["provider"],
            description=payload.get(
                "description", f"{payload['provider']} payments capability"
            ),
            integration_focus=payload.get("integration_focus", []),
            docs=payload.get("docs", []),
            sdks=payload.get("sdks", []),
            setup=payload.get("setup", []),
            troubleshooting=payload.get("troubleshooting", []),
            codefix=payload.get("codefix", []),
            marketing_plays=payload.get("marketing_plays", []),
            keywords=[keyword.lower() for keyword in payload.get("keywords", [])],
            base_maturity=payload.get("base_maturity", "foundational"),
            source=payload.get("source", "payment_platforms_dataset"),
        )


@dataclass(slots=True)
class PaymentIntegrationPlan:
    """Plan describing how to operationalize and troubleshoot payment providers."""

    provider: str
    maturity: str
    description: str
    reasoning: str
    integration_focus: List[str]
    docs: List[str]
    sdks: List[str]
    setup: List[str]
    troubleshooting: List[str]
    codefix: List[str]
    marketing_plays: List[str]
    source: str
    notes: str | None = None


@dataclass(slots=True)
class InvestingCapability:
    """Represents an investing platform intelligence module."""

    provider: str
    category: str
    description: str
    asset_classes: List[str]
    onboarding: List[str]
    monetization: List[str]
    risk_controls: List[str]
    tools: List[str]
    keywords: List[str]
    revenue_programs: List[str]
    kpis: List[str]
    base_maturity: str
    source: str

    @classmethod
    def from_dict(cls, payload: Dict[str, Any]) -> "InvestingCapability":
        return cls(
            provider=payload["provider"],
            category=payload.get("category", "investing"),
            description=payload.get(
                "description",
                f"Investing playbook for {payload['provider']}",
            ),
            asset_classes=payload.get("asset_classes", []),
            onboarding=payload.get("onboarding", []),
            monetization=payload.get("monetization", []),
            risk_controls=payload.get("risk_controls", []),
            tools=payload.get("tools", []),
            keywords=[keyword.lower() for keyword in payload.get("keywords", [])],
            revenue_programs=payload.get("revenue_programs", []),
            kpis=payload.get("kpis", []),
            base_maturity=payload.get("base_maturity", "foundational"),
            source=payload.get("source", "investing_platforms_dataset"),
        )


@dataclass(slots=True)
class InvestingPlan:
    """Plan describing how the ANI activates investing platforms for revenue."""

    provider: str
    category: str
    maturity: str
    description: str
    reasoning: str
    asset_classes: List[str]
    monetization: List[str]
    onboarding: List[str]
    tools: List[str]
    risk_controls: List[str]
    revenue_programs: List[str]
    kpis: List[str]
    source: str
    notes: str | None = None


@dataclass(slots=True)
class KnowledgeWorkflowAsset:
    """Represents a knowledge workflow (RAG or KAG) the ANI can activate."""

    name: str
    category: str
    description: str
    trigger_keywords: List[str]
    indexing_steps: List[str]
    retrieval_steps: List[str]
    vector_store: str
    embedding_models: List[str]
    tooling: List[str]
    evaluation: List[str]
    governance: List[str]
    caching: List[str]
    source: str

    @classmethod
    def from_dict(cls, payload: Dict[str, Any]) -> "KnowledgeWorkflowAsset":
        return cls(
            name=payload["name"],
            category=payload["category"].lower(),
            description=payload["description"],
            trigger_keywords=[keyword.lower() for keyword in payload.get("trigger_keywords", [])],
            indexing_steps=payload.get("indexing_steps", []),
            retrieval_steps=payload.get("retrieval_steps", []),
            vector_store=payload.get("vector_store", "vector_store"),
            embedding_models=payload.get("embedding_models", []),
            tooling=payload.get("tooling", []),
            evaluation=payload.get("evaluation", []),
            governance=payload.get("governance", []),
            caching=payload.get("caching", []),
            source=payload.get("source", "knowledge_workflows_dataset"),
        )


@dataclass(slots=True)
class KnowledgeWorkflowPlan:
    """Plan describing how the ANI operationalizes a knowledge workflow."""

    name: str
    category: str
    maturity: str
    description: str
    reasoning: str
    indexing_plan: List[str]
    retrieval_plan: List[str]
    vector_store: str
    embedding_models: List[str]
    tooling: List[str]
    evaluation: List[str]
    governance: List[str]
    caching: List[str]
    source: str


@dataclass(slots=True)
class ExpertProfile:
    """Domain expert that participates in the mixture-of-experts router."""

    name: str
    provider: str
    description: str
    trigger_keywords: List[str]
    specialties: List[str]
    routing_signals: List[str]
    outputs: List[str]
    fallback: str
    metrics: List[str]
    source: str

    @classmethod
    def from_dict(cls, payload: Dict[str, Any]) -> "ExpertProfile":
        return cls(
            name=payload["name"],
            provider=payload.get("provider", "expert"),
            description=payload.get("description", ""),
            trigger_keywords=[keyword.lower() for keyword in payload.get("trigger_keywords", [])],
            specialties=payload.get("specialties", []),
            routing_signals=payload.get("routing_signals", []),
            outputs=payload.get("outputs", []),
            fallback=payload.get("fallback", ""),
            metrics=payload.get("metrics", []),
            source=payload.get("source", "moe_experts_dataset"),
        )


@dataclass(slots=True)
class ExpertActivation:
    """Represents one activated expert in a mixture-of-experts pass."""

    name: str
    provider: str
    role: str
    confidence: float
    reasoning: str
    outputs: List[str]
    metrics: List[str]
    fallback: str
    source: str


@dataclass(slots=True)
class MixtureOfExpertsPlan:
    """Aggregated mixture-of-experts activation summary."""

    router_summary: str
    maturity: str
    activations: List[ExpertActivation]
    evaluation_metrics: List[str]
    notes: str
    source: str


@dataclass(slots=True)
class AgenticLoopStage:
    """One stage of the perception-reasoning-action-reflection loop."""

    stage: str
    objective: str
    actions: List[str]
    tools: List[str]
    signals: List[str]
    outputs: List[str]
    metrics: List[str]


@dataclass(slots=True)
class AgenticLoopPlan:
    """Describes how the ANI runs a continuous agentic loop."""

    name: str
    goal_focus: str
    summary: str
    maturity: str
    reasoning: str
    stages: List[AgenticLoopStage]
    guardrails: List[str]
    source: str


@dataclass(slots=True)
class QuantumScenarioTemplate:
    """Template describing how to stress test one revenue domain with quantum gates."""

    name: str
    domain: str
    objective: str
    gate_sequence: List[str]
    target_probability: float
    stressors: List[str]
    metrics: List[str]
    integration_points: List[str]
    playbook: List[str]
    notes: str | None = None


@dataclass(slots=True)
class QuantumSimulatorCapability:
    """Quantum simulator metadata loaded from the curated dataset."""

    simulator: str
    provider: str
    description: str
    training_focus: List[str]
    evaluation_metrics: List[str]
    orchestration: List[str]
    dependencies: List[str]
    docs: List[str]
    scenarios: List[QuantumScenarioTemplate]
    source: str = "quantum_simulation_catalog"
    notes: str | None = None


@dataclass(slots=True)
class QuantumStressTest:
    """Result of running one quantum stress scenario."""

    name: str
    domain: str
    objective: str
    stressors: List[str]
    metrics: List[str]
    gate_sequence: List[str]
    measurement_probability: float
    stability_index: float
    integration_points: List[str]
    recommended_actions: List[str]
    notes: str | None
    source: str


@dataclass(slots=True)
class QuantumStressTestPlan:
    """Aggregated quantum simulator plan spanning multiple revenue domains."""

    simulator: str
    provider: str
    maturity: str
    description: str
    reasoning: str
    training_focus: List[str]
    stress_tests: List[QuantumStressTest]
    orchestration_steps: List[str]
    metrics: List[str]
    dependencies: List[str]
    docs: List[str]
    coverage_ratio: float
    domain_stability: Dict[str, float]
    source: str
    notes: str | None = None


@dataclass(slots=True)
class StatisticalEvidence:
    """Posterior insight after blending priors with live campaign metrics."""

    feature: str
    feature_type: str
    posterior_roi: float
    posterior_ctr: float
    confidence: float
    lift_vs_benchmark: float
    sample_size: int
    weight: float
    rationale: str


@dataclass(slots=True)
class StatisticalReasoningSummary:
    """Aggregated statistical reasoning powering marketing ANI."""

    expected_roi: float
    expected_revenue: float
    confidence: float
    success_probability: float
    volatility_adjustment: float
    scenario_spread: Dict[str, float]
    driver_confidence: Dict[str, float]
    evidence: List[StatisticalEvidence]
    narrative: str
    source: str = "statistical_reasoning"
    notes: str | None = None


@dataclass(slots=True)
class BudgetProfile:
    """Baseline investment expectations for a marketing channel."""

    channel: str
    keywords: List[str]
    baseline_cpc: float
    baseline_cpl: float
    avg_conversion_rate: float
    avg_roi: float
    recommended_monthly_spend: float


@dataclass(slots=True)
class BudgetAllocation:
    """Recommended allocation for a marketing campaign channel."""

    channel: str
    audience_focus: str
    offer: str
    objective: str
    allocation_usd: float
    allocation_weight: float
    expected_leads: float
    expected_revenue: float
    rationale: str
    source: str
    notes: str | None = None


@dataclass(slots=True)
class BudgetPlan:
    """Portfolio-level budget recommendation summary."""

    total_budget: float
    allocations: List[BudgetAllocation]
    expected_revenue: float
    expected_roi: float
    payback_period_weeks: float
    summary: str


@dataclass(slots=True)
class AutomationTask:
    """Time-boxed task within an automation sprint."""

    start_minute: int
    end_minute: int
    activity: str
    outcome: str
    tooling: List[str]
    kpis: List[str]
    automation_level: str
    source: str


@dataclass(slots=True)
class AutomationSprint:
    """Three-hour automation run composed of sequential tasks."""

    total_minutes: int
    stage: str
    focus: str
    summary: str
    tasks: List[AutomationTask]
    notes: str | None = None
    source: str = "automation_playbooks"


@dataclass(slots=True)
class PerformanceMemoryEntry:
    """Historical marketing learning captured from closed-loop experiments."""

    play_name: str
    stage: str
    audience_focus: str
    channel: str
    tags: List[str]
    summary: str
    wins: List[str]
    losses: List[str]
    next_actions: List[str]
    kpi_focus: List[str]
    confidence: float
    source: str = "performance_memory"
    notes: str | None = None


@dataclass(slots=True)
class LearningLoopRecommendation:
    """Actionable recommendation generated from the learning memory."""

    play_name: str
    stage: str
    audience_focus: str
    channel: str
    learning: str
    wins: List[str]
    losses: List[str]
    next_actions: List[str]
    kpi_focus: List[str]
    confidence: float
    source: str
    notes: str | None = None


@dataclass(slots=True)
class ExcellenceSummary:
    """Composite readiness score describing how robust the ANI recommendations are."""

    excellence_index: float
    maturity_level: str
    headline: str
    strengths: List[str]
    watchouts: List[str]
    process_coverage: Dict[str, float]
    notes: str | None = None


@dataclass(slots=True)
class StrategicDirective:
    """Directive that fuses multiple process signals into a prioritized motion."""

    name: str
    priority: int
    objective: str
    primary_channel: str
    supporting_channels: List[str]
    investment_usd: float | None
    expected_revenue_usd: float | None
    confidence: float
    time_horizon_hours: int
    north_star_metric: str
    supporting_metrics: List[str]
    action_steps: List[str]
    dependencies: List[str]
    risk_mitigations: List[str]
    inspiration: str
    source: str = "ani_orchestration"
    notes: str | None = None


@dataclass(slots=True)
class ExecutivePlaybook:
    """Cross-process action plan that guides autonomous execution."""

    headline: str
    focus: str
    urgency: str
    north_star_metric: str
    total_budget_usd: float | None
    expected_revenue_usd: float | None
    confidence: float
    directives: List[StrategicDirective]
    summary: str
    notes: str | None = None


@dataclass(slots=True)
class ScenarioProjection:
    """Scenario-level forecast for autonomous marketing execution."""

    scenario: str
    expected_revenue: float
    expected_roi: float
    probability: float
    confidence: float
    accelerators: List[str]
    stabilizers: List[str]
    watchouts: List[str]
    source: str = "scenario_planner"
    notes: str | None = None


@dataclass(slots=True)
class ScenarioAnalysis:
    """Aggregated scenario planning summary."""

    headline: str
    summary: str
    projections: List[ScenarioProjection]
    confidence: float
    accelerators: List[str]
    stabilizers: List[str]
    source: str = "scenario_planner"
    notes: str | None = None


class ScenarioPlanningProcess:
    """Synthesizes optimistic, base, and conservative revenue scenarios."""

    REQUIRED_DOMAINS: set[str] = {
        "google_ads",
        "aws",
        "marketing",
        "bitcoin",
        "stockmarket",
        "stripe",
        "cashapp",
        "venmo",
        "paypal",
        "agentic",
    }

    def __init__(self, volatility_floor: float = 0.15) -> None:
        self._volatility_floor = volatility_floor

    def _estimate_budget(
        self, diagnostics: PromptDiagnostics, campaigns: Sequence[CampaignScore]
    ) -> float:
        stage_baseline = {
            "awareness": 18000.0,
            "consideration": 22000.0,
            "conversion": 28000.0,
            "retention": 20000.0,
        }
        base = stage_baseline.get(diagnostics.stage, 22000.0)
        focus = diagnostics.focus.lower()
        if "enterprise" in focus:
            base *= 1.45
        elif "startup" in focus:
            base *= 0.9
        elif "consumer" in focus:
            base *= 0.8

        if diagnostics.urgency == "urgent":
            base *= 1.25
        elif diagnostics.urgency == "time-sensitive":
            base *= 1.1

        avg_roi = mean(score.metrics.expected_roi for score in campaigns)
        base *= max(0.8, min(1.6, avg_roi / 2.0))

        return round(max(7500.0, base), 2)

    @staticmethod
    def _signal_accelerators(
        google_ads: Sequence[GoogleCampaignPerformance] | None,
        cj_affiliate: Sequence[CJAffiliateRecommendation] | None,
        amazon: Sequence[AmazonProductRecommendation] | None,
        ebay: Sequence[EbayRevenuePlay] | None,
        aws: Sequence[AWSRevenueRecommendation] | None,
        stripe: Sequence[StripeMonetizationPlan] | None,
        paypal: Sequence[PayPalMonetizationPlan] | None,
        payments: Sequence[PaymentIntegrationPlan] | None,
        investing: Sequence[InvestingPlan] | None,
        influencers: Sequence[InfluencerRecommendation] | None,
        stocks: Sequence[StockMarketForecast] | None,
        statistics: StatisticalReasoningSummary | None,
        cognitive: Sequence[CognitiveCapabilityPlan] | None,
        agentic: Sequence[AgenticSystemPlan] | None,
        knowledge: Sequence[KnowledgeWorkflowPlan] | None,
        code_automation: Sequence[CodeAutomationPlan] | None,
        language: Sequence[LanguageCapabilityPlan] | None,
        moe: Sequence[MixtureOfExpertsPlan] | None,
        agentic_loops: Sequence[AgenticLoopPlan] | None,
        quantum: Sequence[QuantumStressTestPlan] | None,
        physics_economics: Sequence[PhysicsEconomicsPlan] | None,
    ) -> List[str]:
        accelerators: List[str] = []
        if google_ads:
            roas = mean(perf.roas for perf in google_ads)
            if roas >= 3.5:
                accelerators.append(f"Google Ads ROAS averaging {roas:.2f}x supports paid scale")
        if cj_affiliate:
            revenue = mean(rec.projected_monthly_revenue for rec in cj_affiliate)
            if revenue > 15000:
                accelerators.append(
                    f"CJ affiliate programs forecast ${revenue:,.0f} monthly expansion"
                )
        if amazon:
            conversion = mean(rec.rating for rec in amazon)
            if conversion >= 4.2:
                accelerators.append("Amazon marketplace sentiment reinforces conversion confidence")
        if ebay:
            margin = mean(play.projected_profit_margin for play in ebay)
            velocity = mean(play.sold_per_month for play in ebay)
            if margin >= 0.25 and velocity >= 100:
                accelerators.append(
                    "eBay partner inventory blends high-margin payouts with rapid sell-through"
                )
        if aws:
            roi = mean(rec.projected_roi for rec in aws)
            if roi >= 2.5:
                accelerators.append("AWS growth plays carry multi-service ROI leverage")
        if stripe:
            price = mean(plan.price_usd for plan in stripe)
            if price >= 99:
                accelerators.append("Stripe monetization supports premium pricing uplift")
        if paypal:
            payout_focus = {plan.payout_schedule for plan in paypal}
            if "daily" in payout_focus:
                accelerators.append("PayPal payouts unlocked daily revenue capture velocity")
            elif paypal:
                accelerators.append("PayPal commerce streamlines cashflow-ready offers")
        if payments:
            top_payment = payments[0]
            if top_payment.maturity in {"advanced", "autonomous"}:
                accelerators.append(
                    f"{top_payment.provider} integration running at {top_payment.maturity} maturity shortens monetization time to value"
                )
            else:
                accelerators.append(
                    f"Payment expertise ready via {top_payment.provider} playbooks for campaign launches"
                )
        if investing:
            top_investing = investing[0]
            if top_investing.maturity in {"advanced", "autonomous"}:
                accelerators.append(
                    f"Investing platform {top_investing.provider} operating at {top_investing.maturity} maturity extends asset monetization"
                )
            else:
                accelerators.append(
                    f"Investing playbooks for {top_investing.provider} ready to convert financial intent"
                )
        if influencers:
            avg_roi = mean(rec.projected_roi for rec in influencers)
            avg_reach = mean(rec.estimated_reach for rec in influencers)
            if avg_roi >= 4.0:
                accelerators.append(
                    f"Influencer partnerships forecast {avg_roi:.1f}x ROI with amplified reach"
                )
            elif avg_reach >= 5000:
                accelerators.append(
                    f"Influencer partners extend campaigns to {avg_reach:,.0f}+ engaged prospects"
                )
        if stocks:
            avg_return = mean(forecast.forecast_return_90d for forecast in stocks)
            avg_conf = mean(forecast.confidence for forecast in stocks)
            if avg_return > 0.12 and avg_conf >= 0.55:
                accelerators.append(
                    f"Equity momentum projects {avg_return*100:.0f}% upside supporting budget expansion"
                )
            elif avg_conf >= 0.6:
                accelerators.append("Stock market baselines stabilizing capital access for growth plays")
        if statistics and statistics.success_probability >= 0.75:
            accelerators.append(
                f"Statistical reasoning signals {statistics.success_probability:.0%} success with {statistics.expected_roi:.2f}x ROI"
            )
        if cognitive:
            top_capability = cognitive[0]
            accelerators.append(
                f"Cognitive architecture {top_capability.maturity} on {top_capability.capability} keeps orchestration sharp"
            )
        if language:
            language_plan = language[0]
            accelerators.append(
                f"Language intelligence {language_plan.capability} operating at {language_plan.maturity} maturity strengthens conversational revenue loops"
            )
        if agentic:
            system = agentic[0]
            if system.maturity in {"elite", "autonomous"}:
                accelerators.append(
                    f"Agentic system {system.system_name} operating at {system.maturity} maturity automates multi-tool execution"
                )
            else:
                accelerators.append(
                    f"Agentic system {system.system_name} activates {system.toolchain[0] if system.toolchain else 'orchestration'} for rapid deployment"
                )
        if knowledge:
            rag_count = len([plan for plan in knowledge if plan.category == "rag"])
            kag_count = len([plan for plan in knowledge if plan.category == "kag"])
            if rag_count:
                accelerators.append(
                    f"RAG pipelines ({rag_count}) keep campaign intelligence instantly retrievable."
                )
            if kag_count:
                accelerators.append(
                    f"Knowledge graphs ({kag_count}) enable multi-hop reasoning for scenario confidence."
                )
        if physics_economics:
            top_plan = physics_economics[0]
            accelerators.append(
                f"Physics/economics plan {top_plan.capability} ({top_plan.maturity}) stabilizes complex revenue mechanics."
            )
        if code_automation:
            plan = code_automation[0]
            accelerators.append(
                f"Code automation {plan.capability} ({plan.maturity}) compresses remediation and deployment cycles."
            )
        if moe:
            top_activation = moe[0].activations[0] if moe[0].activations else None
            if top_activation:
                accelerators.append(
                    f"Mixture-of-experts routed to {top_activation.name} ({top_activation.provider}) with {top_activation.confidence:.0%} confidence."
                )
        if agentic_loops:
            loop = agentic_loops[0]
            accelerators.append(
                f"Agentic loop '{loop.name}' ({loop.maturity}) keeps perception→action cycles continuously tuned."
            )
        if quantum:
            plan = quantum[0]
            covered = {test.domain.lower() for test in plan.stress_tests}
            shared = covered & ScenarioPlanningProcess.REQUIRED_DOMAINS