"""Marketing intelligence engine that ranks and scores enumerated campaigns.""" from __future__ import annotations import base64 import importlib import json import math import os import random 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", ) MARKETING_KNOWLEDGE_DATASET = os.path.join( os.path.dirname(__file__), "..", "..", "daemon", "datasets", "marketing_knowledge.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", ) META_LEARNING_DATASET = os.path.join( os.path.dirname(__file__), "..", "..", "daemon", "datasets", "meta_learning_capabilities.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", ) 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] 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 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 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 notes: str | None = None @dataclass(slots=True) class MetaLearningCapability: """Captures adaptive learning capabilities the ANI can operationalize.""" name: str category: str description: str signal_keywords: List[str] learning_modes: List[str] data_sources: List[str] feedback_channels: List[str] tooling: List[str] metrics: List[str] deployment_modes: List[str] base_maturity: str @classmethod def from_dict(cls, payload: Dict[str, Any]) -> "MetaLearningCapability": return cls( name=payload["name"], category=payload.get("category", "meta_learning"), description=payload["description"], signal_keywords=[keyword.lower() for keyword in payload.get("signal_keywords", [])], learning_modes=payload.get("learning_modes", []), data_sources=payload.get("data_sources", []), feedback_channels=payload.get("feedback_channels", []), tooling=payload.get("tooling", []), metrics=payload.get("metrics", []), deployment_modes=payload.get("deployment_modes", []), base_maturity=payload.get("base_maturity", "foundational"), ) @dataclass(slots=True) class MetaLearningPlan: """Activation plan describing how adaptive learning loops stay smart.""" capability: str category: str maturity: str description: str reasoning: str learning_modes: List[str] data_sources: List[str] feedback_channels: List[str] tooling: List[str] metrics: List[str] deployment_modes: List[str] source: 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 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 MarketingKnowledgeSource: """Represents a marketing knowledge repository the ANI can sync.""" name: str category: str description: str topics: List[str] ingestion_steps: List[str] automation: List[str] coverage_assets: List[str] update_frequency: str tooling: List[str] signals: List[str] metrics: List[str] base_maturity: str source: str @classmethod def from_dict(cls, payload: Dict[str, Any]) -> "MarketingKnowledgeSource": return cls( name=payload["name"], category=payload.get("category", "knowledge"), description=payload.get( "description", f"Knowledge repository for {payload['name']}", ), topics=[topic.lower() for topic in payload.get("topics", [])], ingestion_steps=payload.get("ingestion_steps", []), automation=payload.get("automation", []), coverage_assets=payload.get("coverage_assets", []), update_frequency=payload.get("update_frequency", "weekly"), tooling=payload.get("tooling", []), signals=[signal.lower() for signal in payload.get("signals", [])], metrics=payload.get("metrics", []), base_maturity=payload.get("base_maturity", "foundational"), source=payload.get("source", "marketing_knowledge_dataset"), ) @dataclass(slots=True) class MarketingKnowledgePlan: """Plan describing how the ANI automates marketing knowledge ingestion.""" name: str category: str maturity: str description: str reasoning: str ingestion_steps: List[str] automation: List[str] tooling: List[str] metrics: List[str] coverage_assets: List[str] update_frequency: str source: str notes: str | None = None @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 shots_used: int agents_per_node: int nodes_for_one_million: int compression_ratio: float @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] agents_per_node: int nodes_for_one_million: int compression_index: float shots: int continuous_operation: bool 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, marketing_knowledge: Sequence[MarketingKnowledgePlan] | None, code_automation: Sequence[CodeAutomationPlan] | None, language: Sequence[LanguageCapabilityPlan] | None, meta_learning: Sequence[MetaLearningPlan] | 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 marketing_knowledge: top_plan = marketing_knowledge[0] accelerators.append( f"Marketing knowledge sync '{top_plan.name}' ({top_plan.maturity}) refreshes assets {top_plan.update_frequency}." ) if meta_learning: plan = meta_learning[0] accelerators.append( f"Meta-learning {plan.capability} ({plan.maturity}) compounds experiment intelligence." ) 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 accelerators.append( f"Quantum simulator {plan.simulator} validates {len(shared)} revenue domains with stability {mean(t.stability_index for t in plan.stress_tests):.2f}." ) return accelerators @staticmethod def _stabilizers( diagnostics: PromptDiagnostics, budget: BudgetPlan | None, statistics: StatisticalReasoningSummary | None, cognitive: Sequence[CognitiveCapabilityPlan] | None, agentic: Sequence[AgenticSystemPlan] | None, knowledge: Sequence[KnowledgeWorkflowPlan] | None, language: Sequence[LanguageCapabilityPlan] | None, code_automation: Sequence[CodeAutomationPlan] | None, payments: Sequence[PaymentIntegrationPlan] | None, investing: Sequence[InvestingPlan] | None, agentic_loops: Sequence[AgenticLoopPlan] | None, quantum: Sequence[QuantumStressTestPlan] | None, physics_economics: Sequence[PhysicsEconomicsPlan] | None, meta_learning: Sequence[MetaLearningPlan] | None, marketing_knowledge: Sequence[MarketingKnowledgePlan] | None, ) -> List[str]: stabilizers: List[str] = [] if diagnostics.recommended_kpis: stabilizers.append(f"Instrument KPI guardrails around {diagnostics.recommended_kpis[0]}") if budget and budget.allocations: top_channel = budget.allocations[0].channel stabilizers.append(f"Phase spend across {top_channel} with weekly optimization cycles") if statistics: stabilizers.append( f"Recompute statistical priors weekly to maintain {statistics.success_probability:.0%} success guardrails" ) if cognitive: stabilizers.append( f"Audit cognitive capability '{cognitive[0].capability}' to preserve prompt and reasoning quality" ) if language: stabilizers.append( f"Maintain language workflow '{language[0].capability}' so prompts and transcripts stay machine readable" ) if code_automation: stabilizers.append( f"Run post-deploy audits on {code_automation[0].capability} to keep automation safe" ) if payments: stabilizers.append( f"Document payment runbooks for {payments[0].provider} to safeguard go-live troubleshooting" ) if investing: stabilizers.append( f"Publish investing compliance guardrails for {investing[0].provider} campaigns" ) if agentic: stabilizers.append( f"Schedule agentic system '{agentic[0].system_name}' health checks to keep tool calling compliant" ) if knowledge: stabilizers.append("Refresh RAG/KAG indices weekly to maintain knowledge coverage >90%.") if marketing_knowledge: stabilizers.append( f"Maintain {marketing_knowledge[0].name} sync cadence ({marketing_knowledge[0].update_frequency})." ) if agentic_loops: stabilizers.append("Run loop retrospectives every 12 hours to capture learning deltas.") if quantum: stabilizers.append( f"Review quantum stress telemetry from {quantum[0].simulator} before scaling spend." ) if physics_economics: stabilizers.append( f"Audit physics/economics plan {physics_economics[0].capability} cadence to preserve algorithmic fidelity." ) if meta_learning: stabilizers.append( f"Review meta-learning loop '{meta_learning[0].capability}' checkpoints for safe adaptation." ) return stabilizers def analyze( self, prompt: str, diagnostics: PromptDiagnostics, campaigns: Sequence[CampaignScore], *, budget: BudgetPlan | None, 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, crypto: Sequence[CryptoMarketSnapshot] | None, influencers: Sequence[InfluencerRecommendation] | None, stocks: Sequence[StockMarketForecast] | None, statistics: StatisticalReasoningSummary | None, cognitive: Sequence[CognitiveCapabilityPlan] | None, agentic: Sequence[AgenticSystemPlan] | None, knowledge: Sequence[KnowledgeWorkflowPlan] | None, marketing_knowledge: Sequence[MarketingKnowledgePlan] | None, code_automation: Sequence[CodeAutomationPlan] | None, language: Sequence[LanguageCapabilityPlan] | None, meta_learning: Sequence[MetaLearningPlan] | None, moe: Sequence[MixtureOfExpertsPlan] | None, agentic_loops: Sequence[AgenticLoopPlan] | None, quantum: Sequence[QuantumStressTestPlan] | None, physics_economics: Sequence[PhysicsEconomicsPlan] | None, limit: int = 3, ) -> ScenarioAnalysis: if not campaigns: raise ValueError("Scenario planning requires campaign scores") top = list(campaigns[: max(1, limit)]) agentic = list(agentic or []) investing = list(investing or []) meta_learning = list(meta_learning or []) quantum = list(quantum or []) physics_economics = list(physics_economics or []) marketing_knowledge = list(marketing_knowledge or []) avg_roi = mean(score.metrics.expected_roi for score in top) avg_confidence = mean(score.metrics.confidence for score in top) avg_ctr = mean(score.metrics.expected_ctr for score in top) base_budget = ( budget.total_budget if budget and budget.total_budget > 0 else self._estimate_budget(diagnostics, top) ) base_revenue = ( budget.expected_revenue if budget and budget.expected_revenue > 0 else round(base_budget * max(1.1, avg_roi), 2) ) base_roi = round(base_revenue / base_budget, 2) if base_budget else round(avg_roi, 2) if statistics: inferred_budget = ( statistics.expected_revenue / max(statistics.expected_roi, 0.1) if statistics.expected_roi > 0 else base_budget ) base_budget = max(base_budget, inferred_budget) base_revenue = max(base_revenue, statistics.expected_revenue) base_roi = max(base_roi, statistics.expected_roi) volatility = self._volatility_floor if diagnostics.urgency == "urgent": volatility += 0.1 elif diagnostics.urgency == "time-sensitive": volatility += 0.05 if diagnostics.risk_signals: volatility += 0.08 if crypto and any(snapshot.change_24h and abs(snapshot.change_24h) > 5 for snapshot in crypto): volatility += 0.05 if stocks and any(forecast.volatility > 0.3 for forecast in stocks): volatility += 0.04 elif stocks and mean(forecast.volatility for forecast in stocks) < 0.12: volatility -= 0.02 if quantum: quantum_stability = mean(test.stability_index for plan in quantum for test in plan.stress_tests) if quantum_stability < 0.45: volatility += 0.05 elif quantum_stability > 0.6: volatility -= 0.02 if statistics: volatility = max(volatility, statistics.volatility_adjustment) volatility = min(0.5, max(self._volatility_floor, volatility)) optimistic_multiplier = 1 + volatility conservative_multiplier = max(0.45, 1 - volatility) optimistic_revenue = round(base_revenue * optimistic_multiplier, 2) conservative_revenue = round(base_revenue * conservative_multiplier, 2) optimistic_roi = round(base_roi * (1 + volatility / 1.5), 2) conservative_roi = round(base_roi * (1 - volatility / 1.2), 2) optimistic_conf = min(0.99, avg_confidence + (0.12 - volatility / 2)) base_conf = min(0.95, max(0.25, avg_confidence)) conservative_conf = max(0.2, avg_confidence - volatility / 2) optimistic_prob = max(0.18, 0.32 - (volatility - self._volatility_floor) * 0.6) conservative_prob = max(0.22, 0.28 + (volatility - self._volatility_floor) * 0.5) base_prob = max(0.3, 1.0 - optimistic_prob - conservative_prob) if statistics: blended_base = statistics.success_probability * 0.6 + base_prob * 0.4 base_prob = min(0.85, max(0.25, blended_base)) residual = max(0.15, 1.0 - base_prob) share_denominator = max(optimistic_prob + conservative_prob, 0.01) optimistic_share = optimistic_prob / share_denominator optimistic_prob = min(0.6, residual * optimistic_share) conservative_prob = max(0.1, residual - optimistic_prob) total = optimistic_prob + base_prob + conservative_prob optimistic_prob = round(max(0.05, optimistic_prob / total), 2) base_prob = round(max(0.1, base_prob / total), 2) conservative_prob = round(max(0.05, conservative_prob / total), 2) accelerators = self._signal_accelerators( google_ads=google_ads, cj_affiliate=cj_affiliate, amazon=amazon, ebay=ebay, aws=aws, stripe=stripe, paypal=paypal, payments=payments, investing=investing, influencers=influencers, stocks=stocks, statistics=statistics, cognitive=cognitive, agentic=agentic, knowledge=knowledge, marketing_knowledge=marketing_knowledge, code_automation=code_automation, language=language, meta_learning=meta_learning, moe=moe, agentic_loops=agentic_loops, quantum=quantum, physics_economics=physics_economics, ) stabilizers = self._stabilizers( diagnostics, budget, statistics, cognitive, agentic, knowledge, language, code_automation, payments, investing, agentic_loops, quantum, physics_economics, meta_learning, marketing_knowledge, ) if avg_ctr > 0.6: accelerators.append("Benchmark CTR momentum indicates creative resonance") if len(accelerators) < 2: accelerators.append("Cross-process enrichment unlocks multi-channel leverage") if not stabilizers: stabilizers.append("Institute weekly analytics reviews to course-correct") risk_watchouts = diagnostics.risk_signals[:2] if not risk_watchouts: risk_watchouts = ["Monitor acquisition efficiency as spend scales"] projections = [ ScenarioProjection( scenario="optimistic", expected_revenue=optimistic_revenue, expected_roi=optimistic_roi, probability=optimistic_prob, confidence=round(optimistic_conf, 2), accelerators=accelerators, stabilizers=stabilizers, watchouts=["Ensure fulfillment capacity scales with demand"], ), ScenarioProjection( scenario="base", expected_revenue=round(base_revenue, 2), expected_roi=round(base_roi, 2), probability=base_prob, confidence=round(base_conf, 2), accelerators=accelerators, stabilizers=stabilizers, watchouts=risk_watchouts, ), ScenarioProjection( scenario="conservative", expected_revenue=conservative_revenue, expected_roi=round(conservative_roi, 2), probability=round(conservative_prob, 2), confidence=round(conservative_conf, 2), accelerators=accelerators, stabilizers=stabilizers, watchouts=["Reserve contingency budget to reallocate into proven channels"], ), ] primary_channels = {score.campaign.channel for score in top[:2]} summary = ( f"Model optimistic, base, and conservative revenue trajectories for {diagnostics.stage} " f"outcomes targeting {diagnostics.focus}." ) headline = ( f"Scenario coverage for {', '.join(sorted(primary_channels))} anchored in {diagnostics.focus}." ) notes = ( f"Volatility factor {volatility:.2f} derives from urgency={diagnostics.urgency} and risk signals" ) if statistics: notes += ( f" | Statistical reasoning forecasts {statistics.expected_roi:.2f}x ROI with " f"{statistics.success_probability:.0%} success probability" ) if agentic: notes += ( f" | Agentic system {agentic[0].system_name} ({agentic[0].maturity}) orchestrates tool calling" ) if code_automation: notes += ( f" | Code automation {code_automation[0].capability} ({code_automation[0].maturity}) holds deployment guardrails" ) return ScenarioAnalysis( headline=headline, summary=summary, projections=projections, confidence=round(base_conf, 2), accelerators=accelerators, stabilizers=stabilizers, notes=notes, ) class PerformanceMemory: """Loads and scores historical marketing learnings.""" def __init__(self, entries: Sequence[PerformanceMemoryEntry]): if not entries: raise ValueError("PerformanceMemory requires at least one entry") self._entries = list(entries) @classmethod def from_file(cls, path: str = PERFORMANCE_MEMORY_DATASET) -> "PerformanceMemory": with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) entries = [] for record in payload: entries.append( PerformanceMemoryEntry( play_name=record["play_name"], stage=record["stage"], audience_focus=record["audience_focus"], channel=record["channel"], tags=[tag.lower() for tag in record.get("tags", [])], summary=record["summary"], wins=record.get("wins", []), losses=record.get("losses", []), next_actions=record.get("next_actions", []), kpi_focus=[str(kpi) for kpi in record.get("kpi_focus", [])], confidence=float(record.get("confidence", 0.5)), source=record.get("source", "performance_memory"), notes=record.get("notes"), ) ) return cls(entries) @property def entries(self) -> List[PerformanceMemoryEntry]: return list(self._entries) @dataclass(slots=True) class CompetitiveLandscapeEntry: """Competitive signal captured from market monitoring.""" competitor: str channel: str summary: str strengths: List[str] weaknesses: List[str] recommended_play: str share_of_voice: float signal_intensity: float keywords: List[str] maturity: str source: str notes: str | None = None @dataclass(slots=True) class CompetitiveSignal: """Recommendation derived from the competitive landscape dataset.""" competitor: str channel: str summary: str strengths: List[str] weaknesses: List[str] recommended_play: str share_of_voice: float signal_intensity: float threat_level: str focus_alignment: float source: str notes: str | None = None class CompetitiveLandscape: """Loads competitive marketing signals from curated monitoring.""" def __init__(self, entries: Sequence[CompetitiveLandscapeEntry]): if not entries: raise ValueError("CompetitiveLandscape requires at least one entry") self._entries = list(entries) @classmethod def from_file( cls, path: str = COMPETITIVE_LANDSCAPE_DATASET ) -> "CompetitiveLandscape": with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) entries: List[CompetitiveLandscapeEntry] = [] for record in payload: entries.append( CompetitiveLandscapeEntry( competitor=record["competitor"], channel=record["channel"], summary=record["summary"], strengths=list(record.get("strengths", [])), weaknesses=list(record.get("weaknesses", [])), recommended_play=record["recommended_play"], share_of_voice=float(record.get("share_of_voice", 0.0)), signal_intensity=float(record.get("signal_intensity", 0.0)), keywords=[str(token).lower() for token in record.get("keywords", [])], maturity=record.get("maturity", "growth"), source=record.get("source", "competitive_monitor"), notes=record.get("notes"), ) ) return cls(entries) @property def entries(self) -> List[CompetitiveLandscapeEntry]: return list(self._entries) class LearningLoopProcess: """Transforms prompt intelligence into closed-loop learning recommendations.""" def __init__(self, memory: PerformanceMemory | None = None): self._memory = memory or PerformanceMemory.from_file() def recommend( self, prompt: str, diagnostics: "PromptDiagnostics", campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[LearningLoopRecommendation]: tokens = MarketingIntelligenceEngine._keyword_set(prompt) diag_tokens: set[str] = set() for bucket in (diagnostics.opportunities, diagnostics.recommended_kpis): for item in bucket: diag_tokens.update(MarketingIntelligenceEngine._keyword_set(item)) focus_tokens = set(MarketingIntelligenceEngine._keyword_set(diagnostics.focus)) urgency_bonus = 0.6 if diagnostics.urgency != "standard" else 0.0 channels = {score.campaign.channel.lower() for score in campaigns} audiences = {score.campaign.audience.lower() for score in campaigns} scored: List[Tuple[float, PerformanceMemoryEntry]] = [] for entry in self._memory.entries: entry_tags = set(entry.tags) kpi_tokens: set[str] = set() for kpi in entry.kpi_focus: kpi_tokens.update(MarketingIntelligenceEngine._keyword_set(kpi)) combined_tags = entry_tags.union(kpi_tokens) tag_hits = len(combined_tags.intersection(tokens)) diag_hits = len(combined_tags.intersection(diag_tokens)) focus_hits = len(combined_tags.intersection(focus_tokens)) stage_bonus = 1.8 if entry.stage.lower() == diagnostics.stage.lower() else 0.0 channel_bonus = 1.0 if entry.channel.lower() in channels else 0.0 audience_bonus = 0.0 for audience in audiences: if entry.audience_focus.lower() in audience or any(tag in audience for tag in combined_tags): audience_bonus = 0.9 break focus_bonus = 0.7 if focus_hits else 0.0 opportunity_bonus = 0.5 * diag_hits keyword_bonus = 0.4 * tag_hits score = ( entry.confidence + keyword_bonus + opportunity_bonus + focus_bonus + stage_bonus + channel_bonus + audience_bonus + urgency_bonus ) scored.append((score, entry)) ranked = sorted(scored, key=lambda item: item[0], reverse=True)[: max(1, limit)] recommendations: List[LearningLoopRecommendation] = [] for _, entry in ranked: recommendations.append( LearningLoopRecommendation( play_name=entry.play_name, stage=entry.stage, audience_focus=entry.audience_focus, channel=entry.channel, learning=entry.summary, wins=entry.wins, losses=entry.losses, next_actions=entry.next_actions, kpi_focus=entry.kpi_focus, confidence=round(min(0.99, max(0.0, entry.confidence)), 2), source=entry.source, notes=entry.notes, ) ) return recommendations class CompetitiveIntelligenceProcess: """Turns competitive monitoring into counter-moves for the marketing ANI.""" def __init__(self, landscape: CompetitiveLandscape | None = None): self._landscape = landscape or CompetitiveLandscape.from_file() def signals( self, prompt: str, diagnostics: "PromptDiagnostics", campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[CompetitiveSignal]: prompt_tokens = set(MarketingIntelligenceEngine._keyword_set(prompt)) focus_tokens = set(MarketingIntelligenceEngine._keyword_set(diagnostics.focus)) combined_tokens = prompt_tokens.union(focus_tokens) campaign_channels = {score.campaign.channel.lower() for score in campaigns} stage_alignment = { "paid search": "conversion", "paid social": "consideration", "content syndication": "awareness", "outbound sdr": "conversion", "lifecycle automation": "retention", "partner ecosystem": "consideration", } scored: List[Tuple[float, CompetitiveLandscapeEntry, float]] = [] for entry in self._landscape.entries: entry_tokens = set(entry.keywords) keyword_hits = len(entry_tokens.intersection(prompt_tokens)) focus_hits = len(entry_tokens.intersection(focus_tokens)) overlap_hits = len(entry_tokens.intersection(combined_tokens)) share_component = entry.share_of_voice * 1.5 intensity_component = entry.signal_intensity keyword_component = 0.25 * keyword_hits focus_component = 0.35 * focus_hits channel_component = 0.6 if entry.channel.lower() in campaign_channels else 0.0 stage_tag = stage_alignment.get(entry.channel.lower()) stage_component = 0.4 if stage_tag == diagnostics.stage.lower() else 0.0 maturity_component = 0.3 if entry.maturity.lower() in diagnostics.focus.lower() else 0.0 total_score = ( share_component + intensity_component + keyword_component + focus_component + channel_component + stage_component + maturity_component ) focus_alignment = min(1.0, overlap_hits / max(1, len(entry_tokens))) scored.append((total_score, entry, focus_alignment)) ranked = sorted(scored, key=lambda item: item[0], reverse=True)[: max(1, limit)] signals: List[CompetitiveSignal] = [] for total_score, entry, alignment in ranked: threat_index = entry.share_of_voice * 1.5 + entry.signal_intensity if threat_index >= 1.65: threat_level = "high" elif threat_index >= 1.15: threat_level = "medium" else: threat_level = "low" signals.append( CompetitiveSignal( competitor=entry.competitor, channel=entry.channel, summary=entry.summary, strengths=entry.strengths, weaknesses=entry.weaknesses, recommended_play=entry.recommended_play, share_of_voice=round(entry.share_of_voice, 2), signal_intensity=round(entry.signal_intensity, 2), threat_level=threat_level, focus_alignment=round(alignment, 2), source=entry.source, notes=entry.notes, ) ) return signals class MarketingBenchmarks: """Loads benchmark statistics and produces campaign metrics.""" def __init__(self, baseline: Dict[str, float], components: Dict[str, Dict[str, Dict[str, float]]]): self._baseline = baseline self._components = components @classmethod def from_file(cls, path: str = MARKETING_BENCHMARKS) -> "MarketingBenchmarks": with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) baseline = payload.get("baseline", {"roi": 2.0, "ctr": 0.02, "confidence": 0.6}) components: Dict[str, Dict[str, Dict[str, float]]] = {} for feature, entries in payload.items(): if feature == "baseline": continue components[feature] = {key.lower(): value for key, value in entries.items()} return cls(baseline, components) def _lookup(self, feature: str, value: str) -> Dict[str, float]: cohort = self._components.get(feature, {}) return cohort.get(value.lower(), self._baseline) def metrics_for(self, campaign: MarketingCampaign) -> CampaignMetrics: weights = { "channels": 0.32, "audiences": 0.3, "offers": 0.22, "objectives": 0.16, } baseline_weight = max(0.0, 1.0 - sum(weights.values())) base_roi = self._baseline.get("roi", 2.0) base_ctr = self._baseline.get("ctr", 0.02) base_conf = self._baseline.get("confidence", 0.6) roi = base_roi * baseline_weight ctr = base_ctr * baseline_weight confidence = base_conf * baseline_weight drivers: Dict[str, float] = {} features = { "channels": campaign.channel, "audiences": campaign.audience, "offers": campaign.offer, "objectives": campaign.objective, } for feature, value in features.items(): stats = self._lookup(feature, value) weight = weights[feature] roi += stats.get("roi", base_roi) * weight ctr += stats.get("ctr", base_ctr) * weight confidence += stats.get("confidence", base_conf) * weight drivers[feature.rstrip("s")] = stats.get("roi", base_roi) return CampaignMetrics( expected_roi=roi, expected_ctr=ctr, confidence=min(0.99, max(0.01, confidence)), drivers=drivers, ) @dataclass(slots=True) class CampaignScore: """Composite score for a marketing campaign.""" campaign: MarketingCampaign score: float metrics: CampaignMetrics keyword_coverage: Dict[str, int] @dataclass(slots=True) class PromptDiagnostics: """Structured analysis of a marketing prompt.""" stage: str urgency: str focus: str recommended_kpis: List[str] opportunities: List[str] risk_signals: List[str] keyword_summary: Dict[str, List[str]] confidence: float @dataclass(slots=True) class GrowthExperiment: """Experiment plan produced by the Experiment Growth Grid (EGG) process.""" name: str hypothesis: str primary_metric: str secondary_metrics: List[str] playbook: List[str] confidence: float @dataclass(slots=True) class CJAffiliateOffer: """Offer metadata sourced from the Commission Junction catalog.""" program_name: str advertiser_id: str category: str commission_rate: float epc: float seven_day_epc: float network_rank: int payout_terms: str landing_url: str description: str keywords: List[str] @dataclass(slots=True) class CJAffiliateRecommendation: """Recommendation returned by the CJ affiliate process.""" program_name: str advertiser_id: str category: str commission_rate: float epc: float seven_day_epc: float projected_epc: float projected_monthly_revenue: float network_rank: int payout_terms: str landing_url: str rationale: str source: str notes: str | None = None @dataclass(slots=True) class InfluencerPartner: """Influencer metadata captured from curated partner intelligence.""" name: str platform: str niche: str audience_size: int engagement_rate: float avg_cpm: float collab_formats: List[str] regions: List[str] conversion_focus: str keywords: List[str] case_study: str @dataclass(slots=True) class InfluencerRecommendation: """Recommendation returned by the influencer collaboration process.""" name: str platform: str niche: str collab_format: str audience_size: int engagement_rate: float estimated_reach: int projected_cpm: float projected_leads: int projected_revenue: float projected_roi: float rationale: str source: str notes: str | None = None class InfluencerNetworkCatalog: """Loads curated influencer partner records for offline orchestration.""" def __init__(self, partners: Sequence[InfluencerPartner] | None = None) -> None: if partners is not None: self._partners = list(partners) else: with open(INFLUENCER_PARTNERS_DATASET, "r", encoding="utf-8") as handle: data = json.load(handle) self._partners = [InfluencerPartner(**entry) for entry in data] def match_partners( self, tokens: Sequence[str], focus: str | None, limit: int ) -> List[Tuple[float, InfluencerPartner]]: token_set = {token.lower() for token in tokens} focus_tokens = {token.lower() for token in re.findall(r"[a-zA-Z0-9]+", focus or "")} ranked: List[Tuple[float, InfluencerPartner]] = [] for partner in self._partners: haystack = ( f"{partner.name} {partner.platform} {partner.niche} {partner.conversion_focus} " f"{' '.join(partner.keywords)} {partner.case_study}" ).lower() score = 0.0 keyword_hits = 0 for token in token_set: if token in haystack: score += 1.8 keyword_hits += 1 if token in {keyword.lower() for keyword in partner.keywords}: score += 1.5 keyword_hits += 1 for token in focus_tokens: if token and token in haystack: score += 1.2 if focus and focus.lower() in partner.conversion_focus.lower(): score += 1.0 if focus and focus.lower() in partner.niche.lower(): score += 0.8 score += min(partner.engagement_rate * 20, 2.4) score += min(partner.audience_size / 150000.0, 2.0) score += min(keyword_hits * 0.2, 1.5) if "ai" in token_set and "ai" in {k.lower() for k in partner.keywords}: score += 0.6 ranked.append((score, partner)) ranked.sort(key=lambda item: item[0], reverse=True) return ranked[: max(1, limit)] @dataclass(slots=True) class AmazonMarketplaceListing: """Amazon or AWS Marketplace listing metadata.""" asin: str title: str category: str price_usd: float commission_rate: float rating: float reviews: int url: str keywords: List[str] estimated_monthly_sales: int buy_box_share: float prime_eligible: bool fulfillment_model: str @dataclass(slots=True) class AmazonProductRecommendation: """Recommendation payload produced by the Amazon marketplace process.""" asin: str title: str category: str price_usd: float commission_rate: float rating: float reviews: int estimated_monthly_sales: int buy_box_share: float prime_eligible: bool fulfillment_model: str projected_monthly_revenue: float rationale: str url: str source: str notes: str | None = None class AmazonMarketplaceCatalog: """Loads curated Amazon and AWS marketplace listings for offline use.""" def __init__(self, listings: Sequence[AmazonMarketplaceListing] | None = None): if listings is not None: self._listings = list(listings) else: with open(AMAZON_MARKETPLACE_DATASET, "r", encoding="utf-8") as handle: data = json.load(handle) self._listings = [AmazonMarketplaceListing(**entry) for entry in data] def match_listings( self, tokens: Sequence[str], limit: int, focus: str | None = None ) -> List[Tuple[float, AmazonMarketplaceListing]]: ranked: List[Tuple[float, AmazonMarketplaceListing]] = [] token_set = {token.lower() for token in tokens} for listing in self._listings: haystack = f"{listing.title} {listing.category} {' '.join(listing.keywords)}".lower() score = 0.0 keyword_hits = 0 for token in token_set: if token in haystack: score += 1.5 keyword_hits += 1 if token in {keyword.lower() for keyword in listing.keywords}: score += 1.8 if focus: if focus.lower() in listing.category.lower(): score += 1.2 if focus.lower() in haystack: score += 0.8 score += min(listing.rating / 5.0, 1.0) score += min(listing.reviews / 500.0, 1.0) score += min(listing.estimated_monthly_sales / 400.0, 1.2) score += min(listing.buy_box_share, 1.0) if listing.prime_eligible: score += 0.35 if listing.fulfillment_model.lower().startswith("fba"): score += 0.25 score += keyword_hits * 0.1 ranked.append((score, listing)) ranked.sort(key=lambda item: item[0], reverse=True) return ranked[: max(1, limit)] class AmazonMarketplaceProcess: """Transforms marketplace listings into revenue plans with optional SDK enrichment.""" REQUIRED_ENV_VARS = { "AMAZON_PAAPI_ACCESS_KEY": "access_key", "AMAZON_PAAPI_SECRET_KEY": "[credential removed]", "AMAZON_PAAPI_PARTNER_TAG": "partner_tag", } OPTIONAL_ENV_VARS = { "AWS_REGION": "region", "AWS_PROFILE": "profile", } def __init__( self, catalog: AmazonMarketplaceCatalog | None = None, config: Dict[str, Any] | None = None, ) -> None: self._catalog = catalog or AmazonMarketplaceCatalog() self._config = config or self._config_from_env() self._session = self._build_session() def _build_session(self) -> Any | None: if not self.configured or boto3 is None: return None try: # pragma: no cover - requires boto3 dependency return boto3.session.Session( aws_access_key_id=self._config.get("access_key"), aws_secret_access_key=[credential removed]"secret_key"), region_name=self._config.get("region"), profile_name=self._config.get("profile"), ) except Exception: return None @classmethod def _config_from_env(cls) -> Dict[str, Any]: config: Dict[str, Any] = {} for env_var, key in {**cls.REQUIRED_ENV_VARS, **cls.OPTIONAL_ENV_VARS}.items(): value = os.getenv(env_var) if value: config[key] = value return config @property def configured(self) -> bool: required = {key for key in self.REQUIRED_ENV_VARS.values()} return required <= self._config.keys() def listings( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int, ) -> List[AmazonProductRecommendation]: tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower()) focus = None if campaigns: focus = campaigns[0].campaign.channel.split()[0] matches = self._catalog.match_listings(tokens, limit, focus) if not matches: return [] avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.1 avg_ctr = mean([score.metrics.expected_ctr for score in campaigns]) if campaigns else 0.025 avg_confidence = ( mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.55 ) recommendations: List[AmazonProductRecommendation] = [] for match_score, listing in matches: keyword_overlap = sorted( { token for token in tokens if token in {kw.lower() for kw in listing.keywords} or token in listing.title.lower() } ) intensity = max(0.1, min(2.0, match_score / 4.0)) volume_estimate = listing.estimated_monthly_sales * max(0.4, avg_roi / 2.5) buy_box_lift = 0.75 + listing.buy_box_share * 0.6 prime_lift = 1.18 if listing.prime_eligible else 0.9 confidence_lift = 0.85 + avg_confidence * 0.3 intent_multiplier = 0.35 + intensity * 0.25 + avg_ctr * 9.5 projected_revenue = round( listing.price_usd * listing.commission_rate * volume_estimate * buy_box_lift * prime_lift * confidence_lift * intent_multiplier / 100, 2, ) if keyword_overlap: rationale = ( f"Aligns {listing.category.lower()} offer with {', '.join(keyword_overlap)} demand signals." ) else: rationale = ( f"High rating and volume listing positioned for {listing.category.lower()} expansion." ) source = "amazon_paapi" if self._session else "amazon_catalog" notes = None if not self._session: notes = ( "Simulated listing intelligence; supply Amazon PA-API credentials to activate live harvesting." ) recommendations.append( AmazonProductRecommendation( asin=listing.asin, title=listing.title, category=listing.category, price_usd=listing.price_usd, commission_rate=listing.commission_rate, rating=listing.rating, reviews=listing.reviews, estimated_monthly_sales=listing.estimated_monthly_sales, buy_box_share=listing.buy_box_share, prime_eligible=listing.prime_eligible, fulfillment_model=listing.fulfillment_model, projected_monthly_revenue=projected_revenue, rationale=rationale, url=listing.url, source=source, notes=notes, ) ) return recommendations @dataclass(slots=True) class EbayMarketplaceListing: """eBay marketplace listing metadata for revenue planning.""" item_id: str title: str category: str price_usd: float affiliate_rate: float seller_level: str seller_rating: float conversion_rate: float sold_per_month: int shipping_days: int url: str keywords: List[str] @dataclass(slots=True) class EbayRevenuePlay: """Recommendation payload produced by the eBay marketplace process.""" item_id: str title: str category: str price_usd: float affiliate_rate: float seller_level: str seller_rating: float conversion_rate: float sold_per_month: int projected_monthly_revenue: float projected_profit_margin: float shipping_days: int rationale: str url: str source: str notes: str | None = None class EbayMarketplaceCatalog: """Loads curated eBay marketplace listings for offline use.""" def __init__(self, listings: Sequence[EbayMarketplaceListing] | None = None): if listings is not None: self._listings = list(listings) else: with open(EBAY_MARKETPLACE_DATASET, "r", encoding="utf-8") as handle: data = json.load(handle) self._listings = [EbayMarketplaceListing(**entry) for entry in data] def match_listings( self, tokens: Sequence[str], limit: int, focus: str | None = None ) -> List[Tuple[float, EbayMarketplaceListing]]: ranked: List[Tuple[float, EbayMarketplaceListing]] = [] token_set = {token.lower() for token in tokens} for listing in self._listings: haystack = ( f"{listing.title} {listing.category} {' '.join(listing.keywords)}" ).lower() score = 0.0 keyword_hits = 0 for token in token_set: if token in haystack: score += 1.4 keyword_hits += 1 if token in {keyword.lower() for keyword in listing.keywords}: score += 1.6 if focus: focus_lower = focus.lower() if focus_lower in listing.category.lower(): score += 1.0 if focus_lower in haystack: score += 0.6 score += min(listing.seller_rating / 5.0, 1.0) score += min(listing.conversion_rate * 80.0, 1.2) score += min(listing.sold_per_month / 350.0, 1.2) if listing.seller_level.lower().startswith("top"): score += 0.35 if listing.shipping_days <= 3: score += 0.25 score += keyword_hits * 0.1 ranked.append((score, listing)) ranked.sort(key=lambda item: item[0], reverse=True) return ranked[: max(1, limit)] class EbayMarketplaceProcess: """Transforms eBay inventory into performance-driven affiliate plays.""" REQUIRED_ENV_VARS = { "EBAY_APP_ID": "app_id", "EBAY_CERT_ID": "cert_id", "EBAY_DEV_ID": "dev_id", } OPTIONAL_ENV_VARS = { "EBAY_REDIRECT_URI": "redirect_uri", "EBAY_CAMPAIGN_ID": "campaign_id", } def __init__( self, catalog: EbayMarketplaceCatalog | None = None, config: Dict[str, Any] | None = None, ) -> None: self._catalog = catalog or EbayMarketplaceCatalog() self._config = config or self._config_from_env() self._session = self._build_session() def _build_session(self) -> Any | None: if not self.configured or requests is None: return None try: session = requests.Session() session.headers.update({"User-Agent": "ai-brain-marketing-ani/1.0"}) return session except Exception: return None @classmethod def _config_from_env(cls) -> Dict[str, Any]: config: Dict[str, Any] = {} for env_var, key in {**cls.REQUIRED_ENV_VARS, **cls.OPTIONAL_ENV_VARS}.items(): value = os.getenv(env_var) if value: config[key] = value return config @property def configured(self) -> bool: required = {key for key in self.REQUIRED_ENV_VARS.values()} return required <= self._config.keys() def listings( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int, ) -> List[EbayRevenuePlay]: tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower()) focus = None if campaigns: focus = campaigns[0].campaign.channel.split()[0] matches = self._catalog.match_listings(tokens, limit, focus) if not matches: return [] avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.0 avg_confidence = ( mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.5 ) recommendations: List[EbayRevenuePlay] = [] for match_score, listing in matches: keyword_overlap = sorted( { token for token in tokens if token in {kw.lower() for kw in listing.keywords} or token in listing.title.lower() } ) intensity = max(0.2, min(2.2, match_score / 3.8)) velocity = listing.sold_per_month * max(0.45, avg_roi / 2.2) conversion_lift = 0.6 + listing.conversion_rate * 14.0 seller_lift = 0.8 + (listing.seller_rating / 5.0) * 0.4 speed_lift = 1.05 + max(0, (4 - listing.shipping_days)) * 0.04 confidence_lift = 0.8 + avg_confidence * 0.35 projected_revenue = round( listing.price_usd * listing.affiliate_rate * velocity * conversion_lift * seller_lift * speed_lift * confidence_lift * intensity / 100, 2, ) projected_margin = round( min(0.92, listing.affiliate_rate * (conversion_lift * 0.6 + 0.35)), 2, ) if keyword_overlap: rationale = ( f"Targets {', '.join(keyword_overlap)} buyers with proven conversion velocity." ) else: rationale = ( f"Top-rated {listing.category.lower()} seller with accelerated sell-through momentum." ) source = "ebay_partner_api" if self._session else "ebay_catalog" notes = None if not self._session: notes = ( "Offline catalog heuristics; supply eBay developer credentials to activate live sourcing." ) recommendations.append( EbayRevenuePlay( item_id=listing.item_id, title=listing.title, category=listing.category, price_usd=listing.price_usd, affiliate_rate=listing.affiliate_rate, seller_level=listing.seller_level, seller_rating=listing.seller_rating, conversion_rate=listing.conversion_rate, sold_per_month=listing.sold_per_month, projected_monthly_revenue=projected_revenue, projected_profit_margin=projected_margin, shipping_days=listing.shipping_days, rationale=rationale, url=listing.url, source=source, notes=notes, ) ) return recommendations @dataclass(slots=True) class CryptoMarketSnapshot: """Market intelligence payload for a cryptocurrency asset.""" asset: str price_usd: float | None change_24h: float | None market_cap: float | None volume_24h: float | None source: str notes: str | None = None @dataclass(slots=True) class StockMarketRecord: """Static fundamentals for a global stock market or macro basket.""" symbol: str name: str region: str sector_focus: str current_price: float avg_daily_volume: float market_cap: float pe_ratio: float dividend_yield: float volatility_30d: float momentum_score: float macro_signal: str marketing_correlation: float crypto_correlation: float | None = None @dataclass(slots=True) class StockMarketForecast: """Forecast intelligence summarizing statistical outlooks for a market.""" symbol: str name: str region: str focus: str current_price: float trend: str forecast_return_30d: float forecast_return_90d: float forecast_revenue_multiplier: float volatility: float confidence: float signal_summary: str source: str notes: str | None = None class CryptoMarketProcess: """Ingests crypto market data across public APIs with graceful fallbacks.""" ASSET_ALIASES: Dict[str, str] = { "btc": "bitcoin", "bitcoin": "bitcoin", "sats": "bitcoin", "eth": "ethereum", "ethereum": "ethereum", "sol": "solana", "solana": "solana", "matic": "polygon", "polygon": "polygon", "doge": "dogecoin", "dogecoin": "dogecoin", } DEFAULT_ASSETS: List[str] = ["bitcoin", "ethereum", "solana"] FALLBACK_PRICES: Dict[str, Dict[str, Any]] = { "bitcoin": { "price_usd": 68000.0, "change_24h": 2.5, "market_cap": 1.3e12, "volume_24h": 3.1e10, "source": "simulated", "notes": "Use CoinGecko or CoinCap credentials for live BTC spot feeds.", }, "ethereum": { "price_usd": 3500.0, "change_24h": 3.1, "market_cap": 4.2e11, "volume_24h": 1.8e10, "source": "simulated", "notes": "Hook CoinGecko or CoinCap to mirror ETH market structure.", }, "solana": { "price_usd": 160.0, "change_24h": 4.8, "market_cap": 7.2e10, "volume_24h": 4.1e9, "source": "simulated", "notes": "Pair CoinGecko streams for SOL liquidity snapshots.", }, "polygon": { "price_usd": 0.95, "change_24h": 1.2, "market_cap": 8.8e9, "volume_24h": 6.7e8, "source": "simulated", "notes": "Integrate CoinCap assets endpoint for live Polygon metrics.", }, "dogecoin": { "price_usd": 0.18, "change_24h": 5.5, "market_cap": 2.6e10, "volume_24h": 9.2e8, "source": "simulated", "notes": "CoinGecko provides DOGE spot pricing with rate-limited tier.", }, } def __init__( self, session: Any | None = None, timeout: float = 4.0, allow_network: bool = True, ) -> None: self._session = None if allow_network and requests is not None: self._session = session or requests.Session() self._timeout = timeout @property def configured(self) -> bool: return self._session is not None def _normalize_asset(self, token: str) -> str | None: return self.ASSET_ALIASES.get(token.lower()) def _infer_assets(self, prompt: str, limit: int) -> List[str]: tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower()) assets: List[str] = [] for token in tokens: asset = self._normalize_asset(token) if asset and asset not in assets: assets.append(asset) for asset in self.DEFAULT_ASSETS: if asset not in assets: assets.append(asset) return assets[: max(1, limit)] def _request_json( self, url: str, params: Dict[str, Any] | None = None ) -> Dict[str, Any] | List[Any] | None: if not self._session: return None try: # pragma: no cover - requires network response = self._session.get(url, params=params, timeout=self._timeout) response.raise_for_status() return response.json() except Exception: return None def _fetch_coingecko(self, asset: str) -> Dict[str, Any] | None: data = self._request_json( "https://api.coingecko.com/api/v3/simple/price", params={ "ids": asset, "vs_currencies": "usd", "include_24hr_change": "true", "include_market_cap": "true", "include_24hr_vol": "true", }, ) if isinstance(data, dict) and asset in data: entry = data[asset] return { "price_usd": float(entry.get("usd")) if entry.get("usd") is not None else None, "change_24h": float(entry.get("usd_24h_change")) if entry.get("usd_24h_change") is not None else None, "market_cap": float(entry.get("usd_market_cap")) if entry.get("usd_market_cap") is not None else None, "volume_24h": float(entry.get("usd_24h_vol")) if entry.get("usd_24h_vol") is not None else None, "source": "coingecko", } return None def _fetch_coincap(self, asset: str) -> Dict[str, Any] | None: data = self._request_json(f"https://api.coincap.io/v2/assets/{asset}") if isinstance(data, dict) and isinstance(data.get("data"), dict): entry = data["data"] try: price = float(entry.get("priceUsd")) if entry.get("priceUsd") is not None else None change = float(entry.get("changePercent24Hr")) if entry.get("changePercent24Hr") is not None else None market_cap = float(entry.get("marketCapUsd")) if entry.get("marketCapUsd") is not None else None volume = float(entry.get("volumeUsd24Hr")) if entry.get("volumeUsd24Hr") is not None else None except (TypeError, ValueError): price = change = market_cap = volume = None return { "price_usd": price, "change_24h": change, "market_cap": market_cap, "volume_24h": volume, "source": "coincap", } return None def _fetch_blockstream(self) -> str | None: data = self._request_json("https://blockstream.info/api/mempool") if isinstance(data, dict) and {"count", "vsize"} <= data.keys(): return f"Blockstream mempool: {data['count']} txs, {data['vsize']} vsize" return None def _fetch_blockcypher(self, asset: str) -> str | None: if asset != "bitcoin": return None data = self._request_json("https://api.blockcypher.com/v1/btc/main") if isinstance(data, dict) and {"height", "unconfirmed_count"} <= data.keys(): return ( "BlockCypher height " f"{data['height']} with {data['unconfirmed_count']} unconfirmed tx" ) return None def market_data(self, prompt: str, limit: int = 3) -> List[CryptoMarketSnapshot]: assets = self._infer_assets(prompt, limit) snapshots: List[CryptoMarketSnapshot] = [] blockstream_note = self._fetch_blockstream() for asset in assets: payload = self._fetch_coingecko(asset) or self._fetch_coincap(asset) if payload is None: fallback = self.FALLBACK_PRICES.get(asset) or self.FALLBACK_PRICES["bitcoin"] payload = dict(fallback) else: payload = dict(payload) source = payload.pop("source", "simulated") notes = payload.pop("notes", None) chain_note = self._fetch_blockcypher(asset) combined_notes = " | ".join( note for note in [notes, blockstream_note if asset == "bitcoin" else None, chain_note] if note ) snapshots.append( CryptoMarketSnapshot( asset=asset, price_usd=payload.get("price_usd"), change_24h=payload.get("change_24h"), market_cap=payload.get("market_cap"), volume_24h=payload.get("volume_24h"), source=source, notes=combined_notes or None, ) ) return snapshots class StockMarketProcess: """Builds global stock and bitcoin macro forecasts for marketing orchestration.""" def __init__( self, records: Sequence[StockMarketRecord] | None = None, dataset_path: str = STOCK_MARKETS_DATASET, ) -> None: self._records = list(records) if records is not None else self._load_records(dataset_path) @staticmethod def _load_records(path: str) -> List[StockMarketRecord]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except FileNotFoundError: payload = [] except json.JSONDecodeError: payload = [] records: List[StockMarketRecord] = [] for entry in payload if isinstance(payload, list) else []: try: records.append(StockMarketRecord(**entry)) except TypeError: continue if not records: records.append( StockMarketRecord( symbol="SPX", name="S&P 500 Index", region="North America", sector_focus="Multi-sector", current_price=5200.0, avg_daily_volume=2.4e9, market_cap=40.0, pe_ratio=21.0, dividend_yield=0.015, volatility_30d=0.15, momentum_score=0.35, macro_signal="Fallback macro resilience", marketing_correlation=0.7, crypto_correlation=0.4, ) ) return records @staticmethod def _tokenize(text: str) -> List[str]: return re.findall(r"[a-zA-Z0-9]+", text.lower()) def _score_record( self, record: StockMarketRecord, tokens: Sequence[str], campaigns: Sequence[CampaignScore], ) -> float: token_score = 0.0 haystack = f"{record.name} {record.region} {record.sector_focus} {record.symbol}".lower() for token in tokens: if token in haystack: token_score += 0.6 if token in record.sector_focus.lower(): token_score += 0.4 if token in record.region.lower(): token_score += 0.4 avg_roi = mean(score.metrics.expected_roi for score in campaigns) if campaigns else 2.0 avg_confidence = ( mean(score.metrics.confidence for score in campaigns) if campaigns else 0.5 ) momentum_component = record.momentum_score * 2.0 correlation_component = record.marketing_correlation * 1.5 volatility_penalty = (1.0 - record.volatility_30d) * 1.2 roi_alignment = avg_roi * 0.3 confidence_alignment = avg_confidence * 0.5 return ( token_score + momentum_component + correlation_component + volatility_penalty + roi_alignment + confidence_alignment ) @staticmethod def _trend_label(record: StockMarketRecord) -> str: if record.momentum_score >= 0.45: return "surging" if record.momentum_score >= 0.35: return "climbing" if record.momentum_score >= 0.25: return "stabilizing" return "range bound" def forecasts( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int = 3, ) -> List[StockMarketForecast]: limit = max(1, limit) tokens = self._tokenize(prompt) ranked = sorted( self._records, key=lambda record: self._score_record(record, tokens, campaigns), reverse=True, )[:limit] forecasts: List[StockMarketForecast] = [] avg_roi = mean(score.metrics.expected_roi for score in campaigns) if campaigns else 2.0 avg_ctr = mean(score.metrics.expected_ctr for score in campaigns) if campaigns else 0.03 for record in ranked: volatility = max(0.05, min(0.6, record.volatility_30d)) base_return = record.momentum_score * 0.08 + record.marketing_correlation * 0.05 volatility_adjustment = (0.25 - volatility) * 0.12 thirty_day = max(-0.12, min(0.22, base_return + volatility_adjustment)) ninety_day = max( -0.25, min(0.45, thirty_day * 2.1 + record.momentum_score * 0.05 - volatility * 0.1), ) revenue_multiplier = max( 0.6, min(2.8, (avg_roi / 2.0) * (1 + thirty_day) + avg_ctr * 20), ) confidence = max( 0.25, min( 0.95, 0.45 + record.marketing_correlation * 0.25 + (1 - volatility) * 0.15 + record.momentum_score * 0.15 + (record.crypto_correlation or 0.3) * 0.05, ), ) trend = self._trend_label(record) signal_summary = ( f"{record.name} momentum {trend} with {thirty_day*100:.1f}% 30d outlook " f"and revenue multiplier {revenue_multiplier:.2f}x" ) notes_parts = [record.macro_signal] if record.crypto_correlation is not None and record.crypto_correlation > 0.5: notes_parts.append("High crypto correlation supports synchronized BTC hedges") forecasts.append( StockMarketForecast( symbol=record.symbol, name=record.name, region=record.region, focus=record.sector_focus, current_price=round(record.current_price, 2), trend=trend, forecast_return_30d=round(thirty_day, 4), forecast_return_90d=round(ninety_day, 4), forecast_revenue_multiplier=round(revenue_multiplier, 2), volatility=round(volatility, 3), confidence=round(confidence, 2), signal_summary=signal_summary, source="stock_catalog", notes=" | ".join(notes_parts), ) ) return forecasts class StatisticalReasoningStore: """Optional Postgres sink for statistical reasoning summaries.""" def __init__(self, enabled: bool | None = None) -> None: env_flag = os.getenv("STATISTICAL_REASONING_PERSIST", "0").lower() self._enabled = ( enabled if enabled is not None else env_flag in {"1", "true", "yes", "on"} ) self._config = { "dbname": os.getenv("PGDATABASE", os.getenv("DB_NAME", "brain_atomic")), "user": os.getenv("PGUSER", os.getenv("DB_USER", "postgres")), "host": os.getenv("PGHOST", os.getenv("DB_HOST", "localhost")), "port": int(os.getenv("PGPORT", os.getenv("DB_PORT", "5432"))), } password = os.getenv("PGPASSWORD", os.getenv("DB_PASSWORD")) if password: self._config["password"] = password @property def enabled(self) -> bool: return self._enabled and psycopg2 is not None def record(self, prompt: str, summary: StatisticalReasoningSummary) -> bool: if not self.enabled: return False conn = None cur = None try: # pragma: no cover - requires live Postgres conn = psycopg2.connect(**self._config) # type: ignore[arg-type] cur = conn.cursor() if Json is not None: spread_payload: Any = Json(summary.scenario_spread) # type: ignore[call-arg] drivers_payload: Any = Json(summary.driver_confidence) # type: ignore[call-arg] query = ( "INSERT INTO statistical_reasoning_insights (prompt, expected_roi, expected_revenue, " "confidence, success_probability, volatility_adjustment, scenario_spread, drivers, narrative, notes) " "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)" ) params = ( prompt, summary.expected_roi, summary.expected_revenue, summary.confidence, summary.success_probability, summary.volatility_adjustment, spread_payload, drivers_payload, summary.narrative, summary.notes, ) else: spread_payload = json.dumps(summary.scenario_spread) drivers_payload = json.dumps(summary.driver_confidence) query = ( "INSERT INTO statistical_reasoning_insights (prompt, expected_roi, expected_revenue, " "confidence, success_probability, volatility_adjustment, scenario_spread, drivers, narrative, notes) " "VALUES (%s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s, %s)" ) params = ( prompt, summary.expected_roi, summary.expected_revenue, summary.confidence, summary.success_probability, summary.volatility_adjustment, spread_payload, drivers_payload, summary.narrative, summary.notes, ) cur.execute(query, params) conn.commit() return True except Exception: # pragma: no cover - persistence best-effort return False finally: if cur is not None: cur.close() if conn is not None: conn.close() class StatisticalReasoningProcess: """Blends statistical priors with live campaign metrics to guide revenue.""" def __init__( self, priors: Sequence[StatisticalPrior] | None = None, dataset_path: str = STATISTICAL_PRIORS_DATASET, store: StatisticalReasoningStore | None = None, ) -> None: self._priors = list(priors) if priors is not None else self._load_priors(dataset_path) self._store = store or StatisticalReasoningStore() @staticmethod def _load_priors(path: str) -> List[StatisticalPrior]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] priors: List[StatisticalPrior] = [] if isinstance(payload, list): for entry in payload: if not isinstance(entry, dict): continue try: priors.append( StatisticalPrior( feature=str(entry.get("feature", "unspecified")), feature_type=str(entry.get("feature_type", "feature")), mean_roi=float(entry.get("mean_roi", 2.0)), std_roi=max(0.01, float(entry.get("std_roi", 0.5))), mean_ctr=float(entry.get("mean_ctr", 0.03)), std_ctr=max(0.0001, float(entry.get("std_ctr", 0.01))), sample_size=int(entry.get("sample_size", 100)), weight=float(entry.get("weight", 0.75)), baseline_budget=float(entry.get("baseline_budget", 20000.0)), notes=str(entry.get("notes", "Statistical prior")), ) ) except (TypeError, ValueError): continue if not priors: priors.append( StatisticalPrior( feature="omni-channel", feature_type="channel", mean_roi=2.2, std_roi=0.5, mean_ctr=0.032, std_ctr=0.01, sample_size=120, weight=0.75, baseline_budget=20000.0, notes="Fallback statistical prior", ) ) return priors @staticmethod def _normalize(text: str) -> str: return re.sub(r"\s+", " ", text.lower().strip()) def _match_priors(self, value: str, feature_type: str) -> List[StatisticalPrior]: target = self._normalize(value) matches: List[StatisticalPrior] = [] for prior in self._priors: if prior.feature_type != feature_type: continue feature = self._normalize(prior.feature) if target == feature or target in feature or feature in target: matches.append(prior) return matches def analyze( self, prompt: str, campaigns: Sequence["CampaignScore"], *, budget: BudgetPlan | None, stock_markets: Sequence[StockMarketForecast] | None, crypto_markets: Sequence[CryptoMarketSnapshot] | None, limit: int = 3, persist: bool = True, ) -> StatisticalReasoningSummary: if not campaigns: raise ValueError("Statistical reasoning requires campaign scores") top = list(campaigns[: max(1, limit)]) evidence: List[StatisticalEvidence] = [] driver_confidence: Dict[str, float] = {} baseline_budgets: List[float] = [] for score in top: feature_map = { "channel": score.campaign.channel, "audience": score.campaign.audience, "offer": score.campaign.offer, "objective": score.campaign.objective, } for feature_type, raw_value in feature_map.items(): priors = self._match_priors(raw_value, feature_type) if not priors: continue prior = priors[0] baseline_budgets.append(prior.baseline_budget) prior_weight = max(1.0, prior.sample_size * prior.weight) observation_weight = max(1.0, score.metrics.confidence * 120.0) posterior_roi = ( prior.mean_roi * prior_weight + score.metrics.expected_roi * observation_weight ) / (prior_weight + observation_weight) posterior_ctr = ( prior.mean_ctr * prior_weight + score.metrics.expected_ctr * observation_weight ) / (prior_weight + observation_weight) lift = (posterior_roi - prior.mean_roi) / max(0.1, prior.mean_roi) confidence = max( 0.35, min( 0.98, score.metrics.confidence * 0.6 + prior.weight * 0.4, ), ) rationale = ( f"Posterior {posterior_roi:.2f}x ROI blends prior ({prior.mean_roi:.2f}x) " f"with observed {score.metrics.expected_roi:.2f}x and {prior.sample_size} runs." ) evidence.append( StatisticalEvidence( feature=prior.feature, feature_type=prior.feature_type, posterior_roi=round(posterior_roi, 3), posterior_ctr=round(posterior_ctr, 4), confidence=round(confidence, 3), lift_vs_benchmark=round(lift, 3), sample_size=prior.sample_size, weight=round(prior.weight, 3), rationale=rationale, ) ) driver_key = f"{prior.feature_type}:{prior.feature}" driver_confidence[driver_key] = max( driver_confidence.get(driver_key, 0.0), confidence ) if not evidence: fallback_prior = self._priors[0] evidence.append( StatisticalEvidence( feature=fallback_prior.feature, feature_type=fallback_prior.feature_type, posterior_roi=round(fallback_prior.mean_roi, 3), posterior_ctr=round(fallback_prior.mean_ctr, 4), confidence=0.55, lift_vs_benchmark=0.0, sample_size=fallback_prior.sample_size, weight=round(fallback_prior.weight, 3), rationale="Fallback statistical prior applied due to limited matches.", ) ) driver_confidence[f"{fallback_prior.feature_type}:{fallback_prior.feature}"] = 0.55 baseline_budgets.append(fallback_prior.baseline_budget) total_weight = sum(ev.weight for ev in evidence) or 1.0 expected_roi = sum(ev.posterior_roi * ev.weight for ev in evidence) / total_weight expected_ctr = sum(ev.posterior_ctr * ev.weight for ev in evidence) / total_weight evidence_confidence = sum(ev.confidence for ev in evidence) / len(evidence) avg_confidence = mean(score.metrics.confidence for score in top) roi_signal = mean(score.metrics.expected_roi for score in top) budget_baseline = ( budget.total_budget if budget and budget.total_budget > 0 else (sum(baseline_budgets) / len(baseline_budgets) if baseline_budgets else 22000.0) ) expected_revenue = round(budget_baseline * max(expected_roi, roi_signal, 1.0), 2) stock_vol = ( mean(forecast.volatility for forecast in stock_markets) if stock_markets else 0.22 ) crypto_vol = ( mean(abs(snapshot.change_24h or 0) / 100 for snapshot in crypto_markets) if crypto_markets else 0.12 ) volatility_adjustment = round(min(0.4, stock_vol * 0.6 + crypto_vol * 0.4), 3) confidence = max( 0.35, min( 0.98, evidence_confidence * 0.5 + avg_confidence * 0.3 + min(0.25, expected_ctr * 5) - volatility_adjustment * 0.35, ), ) success_probability = max( 0.15, min( 0.99, 0.5 + (expected_roi - 1.0) * 0.18 + (confidence - 0.5) * 0.5 - volatility_adjustment * 0.25, ), ) scenario_spread = { "optimistic": round(expected_roi * (1 + volatility_adjustment * 0.4 + 0.2), 2), "base": round(expected_roi, 2), "conservative": round(max(0.2, expected_roi * (1 - (volatility_adjustment * 0.5 + 0.15))), 2), } sorted_drivers = sorted( driver_confidence.items(), key=lambda item: item[1], reverse=True ) top_driver_text = ( sorted_drivers[0][0].split(":", 1)[1] if sorted_drivers else "statistical priors" ) narrative = ( f"Statistical reasoning projects {expected_roi:.2f}x ROI with {success_probability:.0%} " f"success probability anchored by {top_driver_text}." ) if stock_markets: narrative += ( f" Market volatility adjustment applied at {volatility_adjustment:.2f} to reflect equity signals." ) driver_confidence = { key: round(min(1.0, max(0.0, value)), 3) for key, value in driver_confidence.items() } notes = ( f"Blended {len(evidence)} statistical priors across {len(top)} campaigns; " f"baseline budget {budget_baseline:,.0f}." ) summary = StatisticalReasoningSummary( expected_roi=round(expected_roi, 3), expected_revenue=expected_revenue, confidence=round(confidence, 3), success_probability=round(success_probability, 3), volatility_adjustment=volatility_adjustment, scenario_spread=scenario_spread, driver_confidence=driver_confidence, evidence=evidence, narrative=narrative, notes=notes, ) if persist: persisted = self._store.record(prompt, summary) if persisted: existing = summary.notes or "" note_parts = [part for part in [existing, "Persisted to Postgres"] if part] summary.notes = " | ".join(note_parts) return summary class CognitiveArchitectureProcess: """Infers cognitive capabilities that power the marketing ANI.""" def __init__( self, capabilities: Sequence[CognitiveCapability] | None = None, dataset_path: str = COGNITIVE_CAPABILITIES_DATASET, ) -> None: self._capabilities = list(capabilities) if capabilities is not None else self._load(dataset_path) @staticmethod def _load(path: str) -> List[CognitiveCapability]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] catalog: List[CognitiveCapability] = [] if isinstance(payload, list): for entry in payload: if isinstance(entry, dict): try: catalog.append(CognitiveCapability.from_dict(entry)) except (KeyError, TypeError, ValueError): continue if not catalog: catalog.append( CognitiveCapability( name="Foundational Cognitive Loop", pillar="foundational", description="Default cognitive layer ensuring reasoning and learning remain available.", signal_keywords=["cognitive", "learning", "reasoning"], prompt_strategies=[ "Request blended learning and reasoning prompts to keep orchestration robust." ], frameworks=["Knowledge-Augmented Generation"], tooling=["Vector memory"], metrics=["excellence_index"], knowledge_assets=["campaign_briefs"], play="Maintain base cognitive readiness for marketing ANI.", ) ) return catalog @staticmethod def _tokenize(text: str) -> List[str]: return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2] def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[CognitiveCapabilityPlan]: if not campaigns: raise ValueError("Cognitive architecture plans require campaign scores") prompt_lower = prompt.lower() prompt_tokens = set(self._tokenize(prompt)) campaign_corpus = " ".join(score.campaign.raw for score in campaigns).lower() campaign_tokens = set(self._tokenize(campaign_corpus)) avg_roi = mean(score.metrics.expected_roi for score in campaigns) avg_confidence = mean(score.metrics.confidence for score in campaigns) plans: List[Tuple[float, CognitiveCapabilityPlan]] = [] for capability in self._capabilities: keyword_hits = [kw for kw in capability.signal_keywords if kw in prompt_lower] campaign_hits = [kw for kw in capability.signal_keywords if kw in campaign_corpus] framework_hits = [ framework for framework in capability.frameworks if framework.lower() in prompt_lower ] score = 0.35 + 0.12 * len(keyword_hits) + 0.08 * len(campaign_hits) + 0.07 * len(framework_hits) if capability.pillar == "machine_learning" and avg_roi >= 2.5: score += 0.12 if capability.pillar == "reasoning" and avg_confidence >= 0.65: score += 0.1 if capability.pillar in {"perception", "vector_databases"} and campaign_tokens & {"amazon", "google", "stock", "crypto"}: score += 0.09 if capability.pillar in {"prompt_engineering", "prompt_management", "language"} and "prompt" in prompt_tokens: score += 0.08 maturity: str if score >= 0.95: maturity = "autonomous" elif score >= 0.7: maturity = "advanced" else: maturity = "foundational" reasoning_parts: List[str] = [] if keyword_hits: reasoning_parts.append( f"Prompt references {', '.join(keyword_hits)} which align to {capability.name}." ) if framework_hits: reasoning_parts.append( f"Frameworks {', '.join(framework_hits)} reinforce the capability stack." ) if not reasoning_parts: reasoning_parts.append( f"Campaign metrics ({avg_roi:.2f}x ROI, {avg_confidence:.0%} confidence) warrant activating {capability.name}." ) reasoning = " ".join(reasoning_parts) prompt_strategy = capability.prompt_strategies[0] if campaigns and any( placeholder in prompt_strategy for placeholder in ["{audience}", "{channel}", "{offer}"] ): prompt_strategy = prompt_strategy.format( audience=campaigns[0].campaign.audience, channel=campaigns[0].campaign.channel, offer=campaigns[0].campaign.offer, ) orchestration_steps = [capability.play] if framework_hits: orchestration_steps.append( f"Accelerate using {', '.join(framework_hits)} for immediate deployment." ) else: orchestration_steps.append( f"Pair with {capability.frameworks[0]} to operationalize within the ANI." ) orchestration_steps.append( "Measure impact via " + ", ".join(capability.metrics[:2]) + "." ) plans.append( ( score, CognitiveCapabilityPlan( capability=capability.name, pillar=capability.pillar, maturity=maturity, description=capability.description, reasoning=reasoning, prompt_strategy=prompt_strategy, frameworks=capability.frameworks, tooling=capability.tooling, metrics=capability.metrics, knowledge_assets=capability.knowledge_assets, orchestration_steps=orchestration_steps, source="cognitive_capability_catalog", ), ) ) plans.sort(key=lambda item: item[0], reverse=True) top_plans = [plan for _, plan in plans[: max(1, limit)]] return top_plans class AgenticSystemsProcess: """Plans agentic system activation across marketing, data, and trading workflows.""" OPTIONAL_ENV_VARS = { "DATABASE_URL": "database_url", "STATISTICS_DATABASE_URL": "statistics_db", "FEATURE_STORE_URL": "feature_store", "VECTOR_DB_URL": "vector_db", "PINECONE_API_KEY": "[credential removed]", "BIGQUERY_PROJECT": "bigquery_project", "AWS_REGION": "aws_region", "BEDROCK_ENDPOINT": "prompt_orchestrator", } def __init__( self, capabilities: Sequence[AgenticCapability] | None = None, dataset_path: str = AGENTIC_SYSTEMS_DATASET, config: Dict[str, Any] | None = None, ) -> None: self._capabilities = list(capabilities) if capabilities is not None else self._load(dataset_path) self._config = config or self._config_from_env() @classmethod def _config_from_env(cls) -> Dict[str, Any]: config: Dict[str, Any] = {} for env_var, key in cls.OPTIONAL_ENV_VARS.items(): value = os.getenv(env_var) if value: config.setdefault(key, value) return config @property def configured(self) -> bool: return bool(self._config) @staticmethod def _load(path: str) -> List[AgenticCapability]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] catalog: List[AgenticCapability] = [] if isinstance(payload, list): for entry in payload: if isinstance(entry, dict): try: catalog.append(AgenticCapability.from_dict(entry)) except (KeyError, TypeError, ValueError): continue if not catalog: catalog.append( AgenticCapability( name="Agentic Growth Skeleton", domain="agentic_systems", description="Baseline agent workflow ensuring orchestration across marketing processes.", controller="LLM executive controller", trigger_keywords=["agent", "automation", "workflow"], frameworks=["Prompt management"], toolchain=["Process orchestrator"], data_interfaces=["Marketing warehouse"], knowledge_sources=["Campaign briefs"], revenue_programs=["Acquisition"], risk_controls=["Budget checkpoints"], maturity_signals=["Automation coverage"], base_maturity="foundational", metrics=["agent_cycle_time"], ) ) return catalog @staticmethod def _tokenize(text: str) -> List[str]: return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2] def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[AgenticSystemPlan]: if not campaigns: raise ValueError("Agentic system plans require campaign scores") prompt_tokens = set(self._tokenize(prompt)) campaign_corpus = " ".join(score.campaign.raw for score in campaigns) campaign_tokens = set(self._tokenize(campaign_corpus)) avg_roi = mean(score.metrics.expected_roi for score in campaigns) avg_confidence = mean(score.metrics.confidence for score in campaigns) config_score = min(0.2, 0.05 * len(self._config)) psycopg2_available = psycopg2 is not None scored_plans: List[Tuple[float, AgenticSystemPlan]] = [] for capability in self._capabilities: keyword_hits = [kw for kw in capability.trigger_keywords if kw in prompt_tokens] campaign_hits = [kw for kw in capability.trigger_keywords if kw in campaign_tokens] framework_hits = [ framework for framework in capability.frameworks if framework.lower() in prompt_tokens ] data_hits = [ interface for interface in capability.data_interfaces if any(keyword in interface.lower() for keyword in prompt_tokens) ] base = 0.45 if capability.base_maturity in {"advanced", "elite"} else 0.35 score = base + 0.09 * len(keyword_hits) + 0.07 * len(campaign_hits) + 0.06 * len(framework_hits) score += 0.05 * len(data_hits) score += config_score if capability.domain in {"machine_learning", "data_connectivity"} and psycopg2_available: score += 0.05 if capability.domain == "market_operations" and any( token in {"trading", "market", "portfolio"} for token in prompt_tokens ): score += 0.06 if capability.domain == "customer_experience" and any( token in {"support", "retention", "customer"} for token in prompt_tokens ): score += 0.05 if capability.domain == "prompt_operations" and {"prompt", "cache", "knowledge"} & prompt_tokens: score += 0.05 if avg_roi >= 2.8: score += 0.05 if avg_confidence >= 0.7: score += 0.04 if score >= 1.05: maturity = "autonomous" elif score >= 0.85: maturity = "elite" elif score >= 0.65: maturity = "advanced" else: maturity = "foundational" if maturity == "foundational" and capability.base_maturity in {"advanced", "elite"}: maturity = capability.base_maturity reasoning_parts = [] if keyword_hits: reasoning_parts.append( f"Prompt signals {', '.join(keyword_hits[:3])} intent aligning to {capability.name}." ) if framework_hits: reasoning_parts.append( f"Framework cues {', '.join(framework_hits[:2])} strengthen orchestration quality." ) if data_hits: reasoning_parts.append( f"Data surfaces ({', '.join(data_hits[:2])}) ensure live telemetry feeds." ) if not reasoning_parts: reasoning_parts.append( f"Campaign ROI {avg_roi:.2f}x and confidence {avg_confidence:.0%} justify activating {capability.name}." ) reasoning = " ".join(reasoning_parts) orchestration_steps = list(capability.toolchain[:3]) orchestration_steps.append( f"Wire {capability.controller} to {capability.frameworks[0] if capability.frameworks else 'agent runner'}" ) orchestration_steps.append("Instrument metrics: " + ", ".join(capability.metrics[:3])) source = "agentic_systems_live" if self.configured else "agentic_systems_catalog" notes = None if not self.configured: notes = ( "Simulated from agentic systems corpus; set DATABASE_URL / FEATURE_STORE_URL / VECTOR_DB_URL " "to enable live tool calling." ) scored_plans.append( ( score, AgenticSystemPlan( system_name=capability.name, domain=capability.domain, controller=capability.controller, maturity=maturity, reasoning=reasoning, orchestration_steps=orchestration_steps, toolchain=list(capability.toolchain), data_interfaces=list(capability.data_interfaces), knowledge_sources=list(capability.knowledge_sources), revenue_programs=list(capability.revenue_programs), risk_controls=list(capability.risk_controls), metrics=list(capability.metrics), source=source, notes=notes, ), ) ) scored_plans.sort(key=lambda item: item[0], reverse=True) return [plan for _, plan in scored_plans[: max(1, limit)]] class LanguageIntelligenceProcess: """Plans how the ANI operationalizes NLP, NLU, NLG, and conversational AI.""" OPTIONAL_ENV_VARS = { "OPENAI_API_KEY": "[credential removed]", "ANTHROPIC_API_KEY": "[credential removed]", "AZURE_OPENAI_ENDPOINT": "llm_providers", "GOOGLE_API_KEY": "[credential removed]", "GOOGLE_APPLICATION_CREDENTIALS": "language_tooling", "AWS_REGION": "aws_region", "AWS_ACCESS_KEY_ID": "aws_access", "AWS_SECRET_ACCESS_KEY": "[credential removed]", } def __init__( self, capabilities: Sequence[LanguageCapability] | None = None, dataset_path: str = LANGUAGE_CAPABILITIES_DATASET, config: Dict[str, Any] | None = None, ) -> None: self._capabilities = ( list(capabilities) if capabilities is not None else self._load(dataset_path) ) self._config = config or self._config_from_env() @classmethod def _config_from_env(cls) -> Dict[str, Any]: config: Dict[str, Any] = {} for env_var, key in cls.OPTIONAL_ENV_VARS.items(): value = os.getenv(env_var) if value: config.setdefault(key, set()).add(env_var.lower()) return {key: sorted(values) for key, values in config.items()} @property def configured(self) -> bool: return bool(self._config) @staticmethod def _load(path: str) -> List[LanguageCapability]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] catalog: List[LanguageCapability] = [] if isinstance(payload, list): for entry in payload: if isinstance(entry, dict): try: catalog.append(LanguageCapability.from_dict(entry)) except (KeyError, TypeError, ValueError): continue if not catalog: catalog.append( LanguageCapability( name="Foundational Language Processing", category="nlp", description="Default NLP layer to tokenize and structure campaign briefs when dataset missing.", signal_keywords=["language", "text", "prompt"], pipeline_components=[ "Tokenization", "Entity extraction", "Sentiment scoring", ], tooling=["spaCy"], datasets=["campaign_briefs"], evaluation_metrics=["precision", "recall"], revenue_programs=["Audience segmentation"], orchestration_play="Stand up baseline NLP preprocessing to unlock downstream orchestration.", ) ) return catalog @staticmethod def _tokenize(text: str) -> List[str]: return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2] def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[LanguageCapabilityPlan]: if not campaigns: raise ValueError("Language plans require campaign scores") prompt_tokens = set(self._tokenize(prompt)) campaign_tokens = set( token for campaign in campaigns for token in self._tokenize( " ".join( [ campaign.campaign.channel, campaign.campaign.audience, campaign.campaign.offer, campaign.campaign.objective, ] ) ) ) avg_confidence = mean(score.metrics.confidence for score in campaigns) plans: List[Tuple[float, LanguageCapabilityPlan]] = [] for capability in self._capabilities: prompt_hits = [kw for kw in capability.signal_keywords if kw in prompt_tokens] campaign_hits = [kw for kw in capability.signal_keywords if kw in campaign_tokens] match_score = len(prompt_hits) * 2 + len(campaign_hits) if any(keyword in prompt.lower() for keyword in capability.signal_keywords): match_score += 1 if capability.category in {"conversational_ai", "nlg"} and avg_confidence < 0.6: match_score += 1 if match_score >= 6 and self.configured: maturity = "autonomous" elif match_score >= 3: maturity = "advanced" else: maturity = "foundational" reasoning_parts = [] if prompt_hits: reasoning_parts.append( f"Prompt references {', '.join(prompt_hits[:3])} signals" ) if campaign_hits: reasoning_parts.append( f"Campaign corpus echoes {', '.join(campaign_hits[:3])} patterns" ) if self.configured: providers = ", ".join(sorted(self._config.keys())) reasoning_parts.append(f"Environment exposes {providers} integrations") if not reasoning_parts: reasoning_parts.append("Default language workflow ensures briefs stay structured") reasoning = "; ".join(reasoning_parts) + "." orchestration_steps = [capability.orchestration_play] orchestration_steps.extend( [f"Operationalize {component.lower()}" for component in capability.pipeline_components[:2]] ) if capability.tooling: orchestration_steps.append( f"Pair with {capability.tooling[0]} to activate production pipeline" ) orchestration_steps.append( "Measure " + ", ".join(capability.evaluation_metrics[:2] or ["precision", "recall"]) + " across live transcripts." ) notes: str | None = None if self._config: providers = sorted( {provider for values in self._config.values() for provider in values} ) notes = "Configured providers: " + ", ".join(providers) plans.append( ( match_score, LanguageCapabilityPlan( capability=capability.name, category=capability.category, maturity=maturity, description=capability.description, reasoning=reasoning, pipeline_components=list(capability.pipeline_components), tooling=list(capability.tooling), datasets=list(capability.datasets), evaluation_metrics=list(capability.evaluation_metrics), revenue_programs=list(capability.revenue_programs), orchestration_steps=orchestration_steps, source="language_capability_catalog", notes=notes, ), ) ) plans.sort(key=lambda item: item[0], reverse=True) return [plan for _, plan in plans[: max(1, limit)]] class CodeAutomationProcess: """Plans how the ANI deploys AI code fixing and intelligent automation.""" OPTIONAL_ENV_VARS = { "GITHUB_TOKEN": "[credential removed]", "GITLAB_TOKEN": "source_control", "BITBUCKET_USERNAME": "source_control", "BITBUCKET_APP_PASSWORD": "[credential removed]", "SONAR_TOKEN": "quality", "SONAR_HOST_URL": "quality", "SONARCLOUD_TOKEN": "quality", "AZURE_DEVOPS_ORG_SERVICE_URL": "pipeline", "AZURE_DEVOPS_EXT_PAT": "pipeline", "CI": "automation", } def __init__( self, capabilities: Sequence[CodeAutomationCapability] | None = None, dataset_path: str = CODE_AUTOMATION_DATASET, config: Dict[str, Any] | None = None, ) -> None: self._capabilities = ( list(capabilities) if capabilities is not None else self._load(dataset_path) ) self._config = config or self._config_from_env() @classmethod def _config_from_env(cls) -> Dict[str, Any]: config: Dict[str, set[str]] = {} for env_var, key in cls.OPTIONAL_ENV_VARS.items(): value = os.getenv(env_var) if value: config.setdefault(key, set()).add(env_var.lower()) return {key: sorted(values) for key, values in config.items()} @property def configured(self) -> bool: return bool(self._config) @staticmethod def _load(path: str) -> List[CodeAutomationCapability]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] catalog: List[CodeAutomationCapability] = [] if isinstance(payload, list): for entry in payload: if isinstance(entry, dict): try: catalog.append(CodeAutomationCapability.from_dict(entry)) except (KeyError, TypeError, ValueError): continue if not catalog: catalog.append( CodeAutomationCapability( name="Baseline Code Automation", category="ai_codefix", description="Fallback automation loop that lint fixes and reruns CI when catalog missing.", signal_keywords=["code", "fix", "automation"], automation_stages=[ "Scan repositories", "Generate remediation plan", "Run unit tests", "Deploy with guardrails", ], tooling=["Static analyzer"], platforms=["GitHub"], metrics=["mean_time_to_repair"], revenue_programs=["Engineering efficiency"], guardrails=["Human review"], orchestration_play="Enable baseline linting and CI automation to unblock releases.", base_maturity="foundational", ) ) return catalog @staticmethod def _tokenize(text: str) -> List[str]: return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2] def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[CodeAutomationPlan]: if not campaigns: raise ValueError("Code automation plans require campaign scores") prompt_tokens = set(self._tokenize(prompt)) campaign_tokens = set( token for campaign in campaigns for token in self._tokenize( " ".join( [ campaign.campaign.channel, campaign.campaign.offer, campaign.campaign.objective, ] ) ) ) avg_roi = mean(score.metrics.expected_roi for score in campaigns) avg_confidence = mean(score.metrics.confidence for score in campaigns) config_bonus = min(0.2, 0.05 * sum(len(values) for values in self._config.values())) plans: List[Tuple[float, CodeAutomationPlan]] = [] for capability in self._capabilities: prompt_hits = [kw for kw in capability.signal_keywords if kw in prompt_tokens] campaign_hits = [kw for kw in capability.signal_keywords if kw in campaign_tokens] automation_hits = [ stage for stage in capability.automation_stages if any(token in stage.lower() for token in prompt_tokens) ] base = 0.45 if capability.base_maturity in {"advanced", "elite"} else 0.35 score = base + 0.09 * len(prompt_hits) + 0.07 * len(campaign_hits) + 0.05 * len(automation_hits) if {"code", "engineering", "dev", "product"} & prompt_tokens and capability.category == "ai_codefix": score += 0.08 if {"automation", "workflow", "process"} & prompt_tokens and capability.category == "intelligent_automation": score += 0.08 if {"agent", "agentic", "autonomous"} & prompt_tokens and capability.category == "agentic_automation": score += 0.08 if avg_roi >= 2.5: score += 0.04 if avg_confidence >= 0.65: score += 0.03 score += config_bonus if score >= 1.05: maturity = "autonomous" elif score >= 0.8: maturity = "advanced" else: maturity = "foundational" base_maturity = capability.base_maturity maturity_order = ["foundational", "advanced", "elite", "autonomous"] if maturity_order.index(maturity) < maturity_order.index(base_maturity): maturity = base_maturity reasoning_parts: List[str] = [] if prompt_hits: reasoning_parts.append( f"Prompt references {', '.join(prompt_hits[:3])} signals tied to {capability.name}." ) if campaign_hits: reasoning_parts.append( f"Campaign context mirrors {', '.join(campaign_hits[:3])} automation focus." ) if automation_hits: reasoning_parts.append( f"Automation stages already align with {', '.join(automation_hits[:2])}." ) if not reasoning_parts: reasoning_parts.append( f"Campaign ROI {avg_roi:.2f}x and confidence {avg_confidence:.0%} justify activating {capability.name}." ) reasoning = " ".join(reasoning_parts) orchestration_steps = [capability.orchestration_play] if capability.automation_stages: orchestration_steps.append( "Stage automation flow: " + " → ".join(capability.automation_stages[:3]) ) if capability.guardrails: orchestration_steps.append( "Guardrails: " + ", ".join(capability.guardrails[:2]) ) orchestration_steps.append( "Track metrics: " + ", ".join(capability.metrics[:2] or ["mean_time_to_repair"]) ) source = "code_automation_configured" if self.configured else "code_automation_catalog" notes = None if not self.configured: notes = ( "Simulated from code automation corpus; set GITHUB_TOKEN/SONAR_TOKEN/AZURE_DEVOPS_EXT_PAT " "to enable live remediation pipelines." ) plans.append( ( score, CodeAutomationPlan( capability=capability.name, category=capability.category, maturity=maturity, description=capability.description, reasoning=reasoning, automation_stages=list(capability.automation_stages), tooling=list(capability.tooling), platforms=list(capability.platforms), metrics=list(capability.metrics), revenue_programs=list(capability.revenue_programs), guardrails=list(capability.guardrails), orchestration_steps=orchestration_steps, source=source, notes=notes, ), ) ) plans.sort(key=lambda item: item[0], reverse=True) return [plan for _, plan in plans[: max(1, limit)]] class MetaLearningProcess: """Scores adaptive learning capabilities that keep the ANI improving.""" OPTIONAL_ENV_VARS = { "RAY_ADDRESS": "reinforcement", "SAGEMAKER_ROLE": "reinforcement", "VERTEX_AI_PROJECT": "meta", "MLFLOW_TRACKING_URI": "distillation", "FEATURE_STORE_URI": "self_supervised", "GOVERNANCE_API_TOKEN": "governance", } def __init__( self, capabilities: Sequence[MetaLearningCapability] | None = None, dataset_path: str = META_LEARNING_DATASET, config: Dict[str, Any] | None = None, ) -> None: self._capabilities = ( list(capabilities) if capabilities is not None else self._load(dataset_path) ) self._config = config or self._config_from_env() @classmethod def _config_from_env(cls) -> Dict[str, List[str]]: config: Dict[str, set[str]] = {} for env_var, key in cls.OPTIONAL_ENV_VARS.items(): if os.getenv(env_var): config.setdefault(key, set()).add(env_var.lower()) return {key: sorted(values) for key, values in config.items()} @property def configured(self) -> bool: return bool(self._config) @staticmethod def _load(path: str) -> List[MetaLearningCapability]: capabilities: List[MetaLearningCapability] = [] try: with open(path, "r", encoding="utf-8") as handle: payloads = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payloads = [] for payload in payloads: try: capabilities.append(MetaLearningCapability.from_dict(payload)) except (KeyError, TypeError, ValueError): continue if not capabilities: capabilities.append( MetaLearningCapability( name="Baseline Learning Loop", category="reinforcement_learning", description="Fallback self-improving loop that tracks experiment results and replays the top performer.", signal_keywords=["learning", "loop", "optimize"], learning_modes=["manual review"], data_sources=["campaign telemetry"], feedback_channels=["weekly audit"], tooling=["spreadsheet"], metrics=["win_rate"], deployment_modes=["manual"], base_maturity="foundational", ) ) return capabilities @staticmethod def _tokenize(text: str) -> List[str]: return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2] def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[MetaLearningPlan]: if not campaigns: raise ValueError("Meta learning plans require campaign scores") prompt_tokens = set(self._tokenize(prompt)) campaign_tokens = set( token for campaign in campaigns for token in self._tokenize( " ".join( [ campaign.campaign.channel, campaign.campaign.offer, campaign.campaign.objective, ] ) ) ) avg_roi = mean(score.metrics.expected_roi for score in campaigns) avg_confidence = mean(score.metrics.confidence for score in campaigns) config_bonus = min(0.12, 0.03 * sum(len(values) for values in self._config.values())) plans: List[Tuple[float, MetaLearningPlan]] = [] for capability in self._capabilities: keyword_hits = [kw for kw in capability.signal_keywords if kw in prompt_tokens] campaign_hits = [kw for kw in capability.signal_keywords if kw in campaign_tokens] mode_hits = [ mode for mode in capability.learning_modes if any(token in mode.lower() for token in prompt_tokens) ] data_hits = [ source for source in capability.data_sources if any(token in source.lower() for token in prompt_tokens) ] base = 0.45 if capability.base_maturity in {"advanced", "elite", "autonomous"} else 0.32 score = base + 0.08 * len(keyword_hits) + 0.05 * len(campaign_hits) + 0.04 * len(mode_hits) score += 0.03 * len(data_hits) if {"learn", "improve", "optimize"} & prompt_tokens: score += 0.06 if avg_roi >= 2.2: score += 0.04 if avg_confidence >= 0.65: score += 0.03 score += config_bonus if score >= 1.05: maturity = "autonomous" elif score >= 0.8: maturity = "advanced" else: maturity = "foundational" maturity_order = ["foundational", "advanced", "elite", "autonomous"] base_index = maturity_order.index(capability.base_maturity) maturity_index = maturity_order.index(maturity) if maturity_index < base_index: maturity = capability.base_maturity reasoning_parts: List[str] = [] if keyword_hits: reasoning_parts.append( f"Prompt references {', '.join(keyword_hits[:3])} signals aligned with {capability.name}." ) if campaign_hits: reasoning_parts.append( f"Campaign mix echoes {', '.join(campaign_hits[:3])} learning themes." ) if mode_hits: reasoning_parts.append( f"Learning modes {', '.join(mode_hits[:2])} already match initiative focus." ) if not reasoning_parts: reasoning_parts.append( f"Average ROI {avg_roi:.2f}x and confidence {avg_confidence:.0%} justify activating {capability.name}." ) orchestration_steps = [ "Learning modes: " + ", ".join(capability.learning_modes[:3] or ["weekly review"]), "Data sources: " + ", ".join(capability.data_sources[:3] or ["campaign telemetry"]), "Feedback channels: " + ", ".join(capability.feedback_channels[:2] or ["manual audit"]), ] source = "meta_learning_configured" if self.configured else "meta_learning_catalog" notes = None if not self.configured: notes = ( "Simulated from meta-learning corpus; set RAY_ADDRESS/SAGEMAKER_ROLE/MLFLOW_TRACKING_URI to enable " "live adaptive learning infrastructure." ) plans.append( ( score, MetaLearningPlan( capability=capability.name, category=capability.category, maturity=maturity, description=capability.description, reasoning=" ".join(reasoning_parts), learning_modes=list(capability.learning_modes), data_sources=list(capability.data_sources), feedback_channels=list(capability.feedback_channels), tooling=list(capability.tooling), metrics=list(capability.metrics), deployment_modes=list(capability.deployment_modes), source=source, notes=notes, ), ) ) plans.sort(key=lambda item: item[0], reverse=True) return [plan for _, plan in plans[: max(1, limit)]] class PhysicsEconomicsProcess: """Blends physics, quantum, and economics readiness into actionable plans.""" OPTIONAL_ENV_VARS = { "QISKIT_RUNTIME": "quantum", "AWS_BRAKET_REGION": "quantum", "GOOGLE_CIRQ_SERVICE_ACCOUNT": "quantum", "ECONOMICS_DATA_WAREHOUSE": "economics", "GPU_RESOURCE_POOL": "hardware", "EDGE_ACCELERATOR_POOL": "hardware", } def __init__( self, capabilities: Sequence[PhysicsEconomicsCapability] | None = None, dataset_path: str = PHYSICS_ECONOMICS_DATASET, config: Dict[str, Any] | None = None, ) -> None: self._capabilities = ( list(capabilities) if capabilities is not None else self._load(dataset_path) ) self._config = config or self._config_from_env() @classmethod def _config_from_env(cls) -> Dict[str, List[str]]: config: Dict[str, set[str]] = {} for env_var, key in cls.OPTIONAL_ENV_VARS.items(): if os.getenv(env_var): config.setdefault(key, set()).add(env_var.lower()) return {key: sorted(values) for key, values in config.items()} @property def configured(self) -> bool: return bool(self._config) @staticmethod def _load(path: str) -> List[PhysicsEconomicsCapability]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] catalog: List[PhysicsEconomicsCapability] = [] if isinstance(payload, list): for entry in payload: if isinstance(entry, dict): try: catalog.append(PhysicsEconomicsCapability.from_dict(entry)) except (KeyError, TypeError, ValueError): continue if not catalog: catalog.append( PhysicsEconomicsCapability( name="Foundational Physics & Economics Readiness", domain="physics_economics", description="Fallback capability that maps marketing prompts to basic physics and economic guardrails.", trigger_keywords=["physics", "economics", "quantum", "pricing", "compression"], marketing_algorithms=["Baseline ROI sensitivity analysis"], physics_principles=["System dynamics balancing"], quantum_strategies=["Monte Carlo variance reduction"], economic_models=["Unit economics calculator"], hardware_strategies=["CPU efficient inference"], data_compression=["Vector sketching"], risk_controls=["Stress test budgets weekly"], metrics=["roi_guardrail"], maturity_signals=["Fallback heuristics"], base_maturity="emerging", orchestration_play="Use fallback heuristics to align marketing spend with baseline physics/economics guardrails.", docs=[], ) ) return catalog @staticmethod def _tokenize(text: str) -> List[str]: return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2] @staticmethod def _phrase_hits(collection: Sequence[str], tokens: Sequence[str]) -> List[str]: hits: List[str] = [] token_set = set(tokens) for phrase in collection: normalized = phrase.lower() if any(token in normalized for token in token_set): hits.append(phrase) return hits def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[PhysicsEconomicsPlan]: if not campaigns: raise ValueError("Physics economics plans require campaign scores") prompt_tokens = set(self._tokenize(prompt)) campaign_tokens = set( token for campaign in campaigns for token in self._tokenize( " ".join( [ campaign.campaign.channel, campaign.campaign.audience, campaign.campaign.offer, campaign.campaign.objective, ] ) ) ) avg_roi = mean(score.metrics.expected_roi for score in campaigns) avg_confidence = mean(score.metrics.confidence for score in campaigns) plans: List[Tuple[float, PhysicsEconomicsPlan]] = [] for capability in self._capabilities: keyword_hits = [ kw for kw in capability.trigger_keywords if kw in prompt_tokens or kw in campaign_tokens ] physics_hits = self._phrase_hits(capability.physics_principles, prompt_tokens) quantum_hits = self._phrase_hits(capability.quantum_strategies, prompt_tokens) economics_hits = self._phrase_hits(capability.economic_models, prompt_tokens) hardware_hits = self._phrase_hits(capability.hardware_strategies, prompt_tokens) compression_hits = self._phrase_hits(capability.data_compression, prompt_tokens) base_score = 0.45 if capability.base_maturity in {"advanced", "elite"} else 0.35 score = base_score score += 0.07 * len(keyword_hits) score += 0.05 * len(physics_hits) score += 0.05 * len(quantum_hits) score += 0.06 * len(economics_hits) score += 0.04 * len(hardware_hits) score += 0.03 * len(compression_hits) if avg_roi >= 2.5: score += 0.05 if avg_confidence >= 0.65: score += 0.04 if {"quantum", "physics", "economics", "pricing"} & prompt_tokens: score += 0.05 if self.configured: score += 0.08 maturity_levels = ["emerging", "advanced", "elite", "autonomous"] base_index = maturity_levels.index(capability.base_maturity) if score >= 1.15: maturity_index = min(base_index + 2, len(maturity_levels) - 1) elif score >= 0.85: maturity_index = min(base_index + 1, len(maturity_levels) - 1) else: maturity_index = base_index maturity = maturity_levels[maturity_index] reasoning_parts: List[str] = [] if keyword_hits: reasoning_parts.append( f"Prompt references {', '.join(keyword_hits[:3])} indicating physics/economics urgency." ) if economics_hits: reasoning_parts.append( f"Economic models {', '.join(economics_hits[:2])} align with revenue objectives." ) if physics_hits or quantum_hits: physics_blend = physics_hits[:2] + quantum_hits[:1] if physics_blend: reasoning_parts.append( f"Physics stack engages {', '.join(physics_blend)} for stress stability." ) if not reasoning_parts: reasoning_parts.append( f"Campaign ROI {avg_roi:.2f}x and confidence {avg_confidence:.0%} justify activating {capability.name}." ) reasoning = " ".join(reasoning_parts) orchestration_steps: List[str] = [capability.orchestration_play] if capability.marketing_algorithms: orchestration_steps.append( "Algorithms: " + ", ".join(capability.marketing_algorithms[:3]) ) if capability.physics_principles: orchestration_steps.append( "Physics focus: " + ", ".join(capability.physics_principles[:2]) ) if capability.economic_models: orchestration_steps.append( "Economic models: " + ", ".join(capability.economic_models[:2]) ) if capability.hardware_strategies: orchestration_steps.append( "Hardware plan: " + ", ".join(capability.hardware_strategies[:2]) ) if capability.data_compression: orchestration_steps.append( "Compression: " + ", ".join(capability.data_compression[:2]) ) if capability.risk_controls: orchestration_steps.append( "Risk controls: " + ", ".join(capability.risk_controls[:2]) ) source = "physics_economics_live" if self.configured else "physics_economics_catalog" notes: str | None = None if not self.configured: notes = ( "Simulated from physics/economics catalog; set QISKIT_RUNTIME / AWS_BRAKET_REGION / ECONOMICS_DATA_WAREHOUSE " "to enable live telemetry." ) plans.append( ( score, PhysicsEconomicsPlan( capability=capability.name, domain=capability.domain, maturity=maturity, description=capability.description, rationale=reasoning, marketing_algorithms=list(capability.marketing_algorithms), physics_principles=list(capability.physics_principles), quantum_strategies=list(capability.quantum_strategies), economic_models=list(capability.economic_models), hardware_strategies=list(capability.hardware_strategies), data_compression=list(capability.data_compression), risk_controls=list(capability.risk_controls), metrics=list(capability.metrics), orchestration_steps=orchestration_steps, docs=list(capability.docs), source=source, notes=notes, ), ) ) plans.sort(key=lambda item: item[0], reverse=True) return [plan for _, plan in plans[: max(1, limit)]] class PaymentIntegrationProcess: """Plans how the ANI orchestrates payment provider integrations.""" PROVIDER_ENV_HINTS: Dict[str, List[str]] = { "stripe": ["STRIPE_API_KEY", "STRIPE_ACCOUNT_ID"], "paypal": ["PAYPAL_CLIENT_ID", "PAYPAL_CLIENT_SECRET", "PAYPAL_MERCHANT_ID"], "cash app": ["SQUARE_ACCESS_TOKEN", "SQUARE_APPLICATION_ID", "SQUARE_LOCATION_ID"], "venmo": ["VENMO_CLIENT_ID", "VENMO_CLIENT_SECRET", "VENMO_ACCESS_TOKEN"], } def __init__( self, capabilities: Sequence[PaymentPlatformCapability] | None = None, dataset_path: str = PAYMENT_PLATFORMS_DATASET, config: Dict[str, List[str]] | None = None, ) -> None: self._capabilities = ( list(capabilities) if capabilities is not None else self._load(dataset_path) ) self._config = config or self._config_from_env() @classmethod def _config_from_env(cls) -> Dict[str, List[str]]: config: Dict[str, List[str]] = {} for provider_key, env_vars in cls.PROVIDER_ENV_HINTS.items(): available = [env for env in env_vars if os.getenv(env)] if available: config[provider_key] = sorted(set(env.lower() for env in available)) return config @property def configured_providers(self) -> set[str]: return set(self._config.keys()) @staticmethod def _load(path: str) -> List[PaymentPlatformCapability]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] catalog: List[PaymentPlatformCapability] = [] if isinstance(payload, list): for entry in payload: if isinstance(entry, dict): try: catalog.append(PaymentPlatformCapability.from_dict(entry)) except (KeyError, TypeError, ValueError): continue if not catalog: catalog.append( PaymentPlatformCapability( provider="Stripe", description="Fallback Stripe guidance for payment readiness.", integration_focus=["Checkout"], docs=["https://docs.stripe.com/"], sdks=["stripe"], setup=["Review Stripe docs"], troubleshooting=["Verify API keys"], codefix=["Regenerate secret keys"], marketing_plays=["Launch payment link"], keywords=["stripe"], base_maturity="foundational", ) ) return catalog @staticmethod def _tokenize(text: str) -> List[str]: return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2] def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 4, ) -> List[PaymentIntegrationPlan]: if not campaigns: raise ValueError("Payment integration plans require campaign scores") prompt_tokens = set(self._tokenize(prompt)) campaign_tokens = set( token for campaign in campaigns for token in self._tokenize( " ".join( [ campaign.campaign.channel, campaign.campaign.audience, campaign.campaign.offer, campaign.campaign.objective, ] ) ) ) plans: List[Tuple[int, PaymentIntegrationPlan]] = [] for capability in self._capabilities: keywords = set(capability.keywords) prompt_hits = sorted(prompt_tokens & keywords) campaign_hits = sorted(campaign_tokens & keywords) score = len(prompt_hits) * 2 + len(campaign_hits) if any(key in prompt.lower() for key in capability.keywords): score += 1 provider_key = capability.provider.lower() configured = False for key, envs in self._config.items(): if provider_key.startswith(key): configured = True provider_key = key break if configured and score >= 3: maturity = "autonomous" elif score >= 3: maturity = "advanced" elif capability.base_maturity: maturity = capability.base_maturity else: maturity = "foundational" reasoning_parts = [] if prompt_hits: reasoning_parts.append( f"Prompt references {', '.join(prompt_hits[:3])} signals" ) if campaign_hits: reasoning_parts.append( f"Campaign corpus echoes {', '.join(campaign_hits[:3])} demand" ) if configured: envs = ", ".join(self._config.get(provider_key, [])) reasoning_parts.append(f"Environment exposes {envs} credentials") if not reasoning_parts: reasoning_parts.append("Fallback payment integration guidance available") notes = None if not configured: hint_envs = self.PROVIDER_ENV_HINTS.get(provider_key, []) if hint_envs: notes = "Export env vars: " + ", ".join(hint_envs) plans.append( ( score, PaymentIntegrationPlan( provider=capability.provider, maturity=maturity, description=capability.description, reasoning="; ".join(reasoning_parts) + ".", integration_focus=list(capability.integration_focus), docs=list(capability.docs), sdks=list(capability.sdks), setup=list(capability.setup), troubleshooting=list(capability.troubleshooting), codefix=list(capability.codefix), marketing_plays=list(capability.marketing_plays), source=capability.source, notes=notes, ), ) ) plans.sort(key=lambda item: item[0], reverse=True) return [plan for _, plan in plans[: max(1, limit)]] class InvestingProcess: """Designs investing platform activation plans.""" def __init__( self, capabilities: Sequence[InvestingCapability] | None = None, dataset_path: str = INVESTING_PLATFORMS_DATASET, ) -> None: self._capabilities = ( list(capabilities) if capabilities is not None else self._load(dataset_path) ) @staticmethod def _load(path: str) -> List[InvestingCapability]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] catalog: List[InvestingCapability] = [] if isinstance(payload, list): for entry in payload: if isinstance(entry, dict): try: catalog.append(InvestingCapability.from_dict(entry)) except (KeyError, TypeError, ValueError): continue if not catalog: catalog.append( InvestingCapability( provider="Fallback Investing", category="investing", description="Baseline investing plan to highlight brokerage monetization paths.", asset_classes=["equities", "etfs"], onboarding=["Publish investing landing page", "Launch nurture emails"], monetization=["Promote brokerage affiliate", "Sell premium research"], risk_controls=["Disclose investing risks", "Monitor compliance copy"], tools=["Broker dashboard"], keywords=["invest", "brokerage", "stock"], revenue_programs=["Broker affiliate"], kpis=["funded_accounts", "aum_contributed"], base_maturity="foundational", source="investing_platforms_fallback", ) ) return catalog @staticmethod def _tokenize(text: str) -> List[str]: return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2] @staticmethod def _phrase_hits(collection: Sequence[str], token_set: set[str]) -> List[str]: hits: List[str] = [] for phrase in collection: normalized = phrase.lower() if any(token in normalized for token in token_set): hits.append(phrase) return hits def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[InvestingPlan]: if not campaigns: raise ValueError("Investing plans require campaign scores") prompt_tokens = set(self._tokenize(prompt)) campaign_tokens = set( token for campaign in campaigns for token in self._tokenize( " ".join( [ campaign.campaign.channel, campaign.campaign.offer, campaign.campaign.objective, campaign.campaign.audience, ] ) ) ) combined_tokens = prompt_tokens | campaign_tokens avg_roi = mean(score.metrics.expected_roi for score in campaigns) avg_confidence = mean(score.metrics.confidence for score in campaigns) plans: List[Tuple[float, InvestingPlan]] = [] for capability in self._capabilities: keyword_hits = sorted(set(capability.keywords) & combined_tokens) asset_hits = self._phrase_hits(capability.asset_classes, combined_tokens) monetization_hits = self._phrase_hits(capability.monetization, combined_tokens) onboarding_hits = self._phrase_hits(capability.onboarding, prompt_tokens) category_hit = capability.category.lower() in combined_tokens score = 0.45 score += 0.1 * len(keyword_hits) score += 0.07 * len(asset_hits) score += 0.06 * len(monetization_hits) score += 0.05 * len(onboarding_hits) if category_hit: score += 0.08 if "invest" in prompt_tokens or "trading" in prompt_tokens: score += 0.05 if avg_roi >= 2.5: score += 0.06 if avg_confidence >= 0.65: score += 0.04 if score >= 1.05: maturity = "autonomous" elif score >= 0.8: maturity = "advanced" else: maturity = "foundational" base_order = ["foundational", "advanced", "autonomous"] if base_order.index(maturity) < base_order.index(capability.base_maturity): maturity = capability.base_maturity reasoning_parts: List[str] = [] if keyword_hits: reasoning_parts.append( f"Prompt and campaigns reference {', '.join(keyword_hits[:3])} investing signals" ) if asset_hits: reasoning_parts.append( f"Asset focus aligns with {', '.join(asset_hits[:3])} plays" ) if monetization_hits: reasoning_parts.append( f"Monetization hooks mention {', '.join(monetization_hits[:2])}" ) if onboarding_hits: reasoning_parts.append( f"Prompt language maps to onboarding step {onboarding_hits[0]}" ) if not reasoning_parts: reasoning_parts.append("Fallback investing monetization playbook ready") plans.append( ( score, InvestingPlan( provider=capability.provider, category=capability.category, maturity=maturity, description=capability.description, reasoning="; ".join(reasoning_parts) + ".", asset_classes=list(capability.asset_classes), monetization=list(capability.monetization), onboarding=list(capability.onboarding), tools=list(capability.tools), risk_controls=list(capability.risk_controls), revenue_programs=list(capability.revenue_programs), kpis=list(capability.kpis), source=capability.source, notes=None, ), ) ) plans.sort(key=lambda item: item[0], reverse=True) return [plan for _, plan in plans[: max(1, limit)]] class KnowledgeWorkflowProcess: """Designs retrieval and knowledge graph plans for the ANI.""" def __init__( self, assets: Sequence[KnowledgeWorkflowAsset] | None = None, dataset_path: str = KNOWLEDGE_WORKFLOWS_DATASET, ) -> None: self._assets = list(assets) if assets is not None else self._load(dataset_path) @staticmethod def _load(path: str) -> List[KnowledgeWorkflowAsset]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] catalog: List[KnowledgeWorkflowAsset] = [] if isinstance(payload, list): for entry in payload: if isinstance(entry, dict): try: catalog.append(KnowledgeWorkflowAsset.from_dict(entry)) except (KeyError, TypeError, ValueError): continue if not catalog: catalog.append( KnowledgeWorkflowAsset( name="Foundational Retrieval Workflow", category="rag", description="Fallback retrieval workflow for campaign collateral.", trigger_keywords=["campaign", "retrieval"], indexing_steps=["Collect campaign documents", "Chunk and embed"], retrieval_steps=["Similarity search", "Augment prompt"], vector_store="faiss", embedding_models=["sentence-transformers/all-MiniLM-L6-v2"], tooling=["LangChain"], evaluation=["context_precision"], governance=["document versioning"], caching=["prompt cache"], source="knowledge_workflows_fallback", ) ) return catalog @staticmethod def _tokenize(text: str) -> set[str]: return {token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2} def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[KnowledgeWorkflowPlan]: if not campaigns: raise ValueError("Knowledge workflow plans require campaign scores") prompt_tokens = self._tokenize(prompt) campaign_corpus = " ".join(score.campaign.raw for score in campaigns) campaign_tokens = self._tokenize(campaign_corpus) avg_confidence = mean(score.metrics.confidence for score in campaigns) avg_roi = mean(score.metrics.expected_roi for score in campaigns) plans_by_category: Dict[str, List[Tuple[float, KnowledgeWorkflowPlan]]] = {} for asset in self._assets: keyword_hits = [kw for kw in asset.trigger_keywords if kw in prompt_tokens] campaign_hits = [kw for kw in asset.trigger_keywords if kw in campaign_tokens] tooling_hits = [tool for tool in asset.tooling if tool.lower() in prompt_tokens] score = 0.45 + 0.12 * len(keyword_hits) + 0.08 * len(campaign_hits) + 0.05 * len(tooling_hits) if asset.category == "rag" and {"rag", "retrieval", "cache"} & prompt_tokens: score += 0.15 if asset.category == "kag" and {"kag", "graph", "knowledge"} & prompt_tokens: score += 0.18 if asset.category == "kag" and avg_confidence >= 0.7: score += 0.07 if asset.category == "rag" and avg_roi >= 2.5: score += 0.07 maturity: str if score >= 1.15: maturity = "autonomous" elif score >= 0.85: maturity = "advanced" else: maturity = "foundational" reasoning_parts: List[str] = [] if keyword_hits: reasoning_parts.append( f"Prompt signals {', '.join(keyword_hits[:3])} which align to {asset.name}." ) if tooling_hits: reasoning_parts.append( f"Tooling cues {', '.join(tooling_hits[:2])} are already referenced in the brief." ) if not reasoning_parts: reasoning_parts.append( f"Campaign ROI {avg_roi:.2f}x and confidence {avg_confidence:.0%} benefit from {asset.category.upper()} orchestration." ) reasoning = " ".join(reasoning_parts) indexing_plan = list(asset.indexing_steps) retrieval_plan = list(asset.retrieval_steps) primary_campaign = campaigns[0].campaign if campaigns else None if primary_campaign: indexing_plan.append( f"Tag chunks with channel={primary_campaign.channel.lower()} and audience={primary_campaign.audience.lower()}" ) retrieval_plan.append( f"Prioritize offers related to {primary_campaign.offer.lower()} during similarity search" ) plan = KnowledgeWorkflowPlan( name=asset.name, category=asset.category, maturity=maturity, description=asset.description, reasoning=reasoning, indexing_plan=indexing_plan, retrieval_plan=retrieval_plan, vector_store=asset.vector_store, embedding_models=list(asset.embedding_models), tooling=list(asset.tooling), evaluation=list(asset.evaluation), governance=list(asset.governance), caching=list(asset.caching), source=asset.source, ) plans_by_category.setdefault(asset.category, []).append((score, plan)) plans: List[KnowledgeWorkflowPlan] = [] for category, items in plans_by_category.items(): items.sort(key=lambda item: item[0], reverse=True) category_limit = max(1, min(limit, len(items))) plans.extend(plan for _, plan in items[:category_limit]) return plans class MarketingKnowledgeProcess: """Automates ingestion of marketing knowledge repositories.""" OPTIONAL_ENV_VARS: Dict[str, str] = { "NOTION_API_KEY": "[credential removed]", "AIRTABLE_API_KEY": "[credential removed]", "CONFLUENCE_API_TOKEN": "confluence", "SLACK_BOT_TOKEN": "slack", "GOOGLE_APPLICATION_CREDENTIALS": "gdrive", "DROPBOX_TOKEN": "dropbox", } def __init__( self, sources: Sequence[MarketingKnowledgeSource] | None = None, dataset_path: str = MARKETING_KNOWLEDGE_DATASET, connectors: Dict[str, List[str]] | None = None, ) -> None: self._sources = list(sources) if sources is not None else self._load(dataset_path) self._connectors = connectors or self._connectors_from_env() @classmethod def _connectors_from_env(cls) -> Dict[str, List[str]]: connectors: Dict[str, set[str]] = {} for env_var, key in cls.OPTIONAL_ENV_VARS.items(): if os.getenv(env_var): connectors.setdefault(key, set()).add(env_var.lower()) return {key: sorted(values) for key, values in connectors.items()} @property def configured(self) -> bool: return bool(self._connectors) @staticmethod def _load(path: str) -> List[MarketingKnowledgeSource]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] sources: List[MarketingKnowledgeSource] = [] if isinstance(payload, list): for entry in payload: if isinstance(entry, dict): try: sources.append(MarketingKnowledgeSource.from_dict(entry)) except (KeyError, TypeError, ValueError): continue if not sources: sources.append( MarketingKnowledgeSource( name="Fallback Marketing Wiki", category="documentation", description="Baseline marketing wiki sync to keep prompts contextualized.", topics=["marketing", "campaign", "playbook"], ingestion_steps=["Export wiki pages", "Chunk and store"], automation=["Nightly export", "Diff changes", "Notify owners"], coverage_assets=["playbooks", "personas", "faqs"], update_frequency="weekly", tooling=["filesystem"], signals=["wiki", "documentation"], metrics=["refresh cadence"], base_maturity="foundational", source="marketing_knowledge_fallback", ) ) return sources @staticmethod def _tokenize(text: str) -> set[str]: return {token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2} def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[MarketingKnowledgePlan]: if not campaigns: raise ValueError("Marketing knowledge plans require campaign scores") prompt_tokens = self._tokenize(prompt) campaign_tokens = self._tokenize(" ".join(score.campaign.raw for score in campaigns)) channel_tokens = { score.campaign.channel.lower().split()[0] for score in campaigns if score.campaign.channel } avg_roi = mean(score.metrics.expected_roi for score in campaigns) avg_confidence = mean(score.metrics.confidence for score in campaigns) connector_bonus = min( 0.15, 0.04 * sum(len(values) for values in self._connectors.values()) ) plans: List[Tuple[float, MarketingKnowledgePlan]] = [] for source in self._sources: topic_hits = [topic for topic in source.topics if topic in prompt_tokens] signal_hits = [signal for signal in source.signals if signal in prompt_tokens] coverage_hits = [ asset for asset in source.coverage_assets if any(token in asset.lower() for token in prompt_tokens | campaign_tokens) or any(asset.lower().startswith(token) for token in channel_tokens) ] tooling_hits = [ tool for tool in source.tooling if tool.lower() in prompt_tokens or tool.lower().split()[0] in prompt_tokens ] connector_hits = [ key for key in self._connectors if key in source.signals or any(key in topic for topic in source.topics) ] base = 0.4 if source.base_maturity in {"advanced", "autonomous"} else 0.32 score = base score += 0.08 * len(topic_hits) score += 0.07 * len(signal_hits) score += 0.05 * len(coverage_hits) score += 0.04 * len(tooling_hits) score += 0.05 * len(connector_hits) if {"knowledge", "wiki", "documentation", "library"} & prompt_tokens: score += 0.05 if avg_roi >= 2.4: score += 0.04 if avg_confidence >= 0.65: score += 0.03 score += connector_bonus if score >= 1.1: maturity = "autonomous" elif score >= 0.82: maturity = "advanced" else: maturity = "foundational" levels = ["foundational", "advanced", "autonomous"] base_index = levels.index(source.base_maturity) maturity_index = levels.index(maturity) if maturity_index < base_index: maturity = source.base_maturity reasoning_parts: List[str] = [] if topic_hits: reasoning_parts.append( f"Prompt references {', '.join(topic_hits[:3])} aligning to {source.name}." ) if coverage_hits: reasoning_parts.append( f"Coverage includes {', '.join(coverage_hits[:3])} needed for campaign reuse." ) if connector_hits: reasoning_parts.append( f"Live connectors detected for {', '.join(connector_hits[:2])} enable automation." ) if not reasoning_parts: reasoning_parts.append( f"Baseline ingestion keeps marketing knowledge fresh for {source.category}." ) reasoning = " ".join(reasoning_parts) missing_envs = [ env for env, key in self.OPTIONAL_ENV_VARS.items() if key in source.signals and key not in self._connectors ] notes = None if missing_envs: notes = ( "Set " + ", ".join(sorted(missing_envs)) + " to enable real-time ingestion for this repository." ) elif not self.configured: notes = "Simulated from marketing knowledge corpus; configure connectors for live sync." plans.append( ( score, MarketingKnowledgePlan( name=source.name, category=source.category, maturity=maturity, description=source.description, reasoning=reasoning, ingestion_steps=list(source.ingestion_steps), automation=list(source.automation), tooling=list(source.tooling), metrics=list(source.metrics), coverage_assets=list(source.coverage_assets), update_frequency=source.update_frequency, source=source.source, notes=notes, ), ) ) plans.sort(key=lambda item: item[0], reverse=True) return [plan for _, plan in plans[: max(1, limit)]] class MixtureOfExpertsProcess: """Routes prompts through marketing monetization experts.""" def __init__( self, experts: Sequence[ExpertProfile] | None = None, dataset_path: str = MOE_EXPERTS_DATASET, ) -> None: self._experts = list(experts) if experts is not None else self._load(dataset_path) @staticmethod def _load(path: str) -> List[ExpertProfile]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] profiles: List[ExpertProfile] = [] if isinstance(payload, list): for entry in payload: if isinstance(entry, dict): try: profiles.append(ExpertProfile.from_dict(entry)) except (KeyError, TypeError, ValueError): continue if not profiles: profiles.append( ExpertProfile( name="Benchmark Revenue Expert", provider="benchmark", description="Fallback expert using benchmark datasets to inform routing.", trigger_keywords=["revenue", "marketing"], specialties=["revenue operations"], routing_signals=["roi"], outputs=["recommendations"], fallback="Use benchmark dataset", metrics=["roi", "ctr"], source="moe_experts_fallback", ) ) return profiles @staticmethod def _tokenize(text: str) -> set[str]: return {token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2} def activate( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 3, ) -> List[MixtureOfExpertsPlan]: if not campaigns: raise ValueError("Mixture-of-experts activation requires campaign scores") prompt_tokens = self._tokenize(prompt) campaign_corpus = " ".join(score.campaign.raw for score in campaigns) campaign_tokens = self._tokenize(campaign_corpus) avg_roi = mean(score.metrics.expected_roi for score in campaigns) avg_confidence = mean(score.metrics.confidence for score in campaigns) scored: List[Tuple[float, ExpertProfile, float]] = [] for expert in self._experts: keyword_hits = [kw for kw in expert.trigger_keywords if kw in prompt_tokens or kw in campaign_tokens] specialty_hits = [ specialty for specialty in expert.specialties if any(word in campaign_corpus.lower() for word in specialty.lower().split()) ] provider_bonus = 0.0 if expert.provider == "google" and {"google", "sem", "ads"} & prompt_tokens: provider_bonus += 0.18 if expert.provider == "stripe" and {"stripe", "pricing", "checkout"} & prompt_tokens: provider_bonus += 0.18 if expert.provider == "cj" and {"affiliate", "cj", "publisher"} & prompt_tokens: provider_bonus += 0.16 if expert.provider == "amazon" and {"amazon", "asin", "prime"} & prompt_tokens: provider_bonus += 0.16 score = 0.42 + 0.14 * len(keyword_hits) + 0.09 * len(specialty_hits) + provider_bonus if avg_roi >= 2.8: score += 0.06 if avg_confidence >= 0.7: score += 0.05 confidence = round(min(0.95, max(0.4, score / 1.2)), 2) scored.append((score, expert, confidence)) scored.sort(key=lambda item: item[0], reverse=True) if not scored: return [] top_k = min(max(2, limit), len(scored)) activations: List[ExpertActivation] = [] metrics: set[str] = set() router_notes: List[str] = [] for score, expert, confidence in scored[:top_k]: reason_parts: List[str] = [] matched = [kw for kw in expert.trigger_keywords if kw in prompt_tokens] if matched: reason_parts.append(f"Router matched tokens {', '.join(matched[:3])}.") if expert.specialties: reason_parts.append(f"Specialty focus: {expert.specialties[0]}.") if not matched: reason_parts.append("Activated for coverage balance across revenue levers.") metrics.update(expert.metrics) router_notes.append(f"{expert.name} confidence {confidence:.0%}") activations.append( ExpertActivation( name=expert.name, provider=expert.provider, role=expert.specialties[0] if expert.specialties else "expert", confidence=confidence, reasoning=" ".join(reason_parts), outputs=list(expert.outputs), metrics=list(expert.metrics), fallback=expert.fallback, source=expert.source, ) ) avg_conf = mean(act.confidence for act in activations) if avg_conf >= 0.8 and len(activations) >= 3: maturity = "autonomous" elif avg_conf >= 0.65: maturity = "advanced" else: maturity = "foundational" summary_tokens = sorted((prompt_tokens & {"google", "stripe", "cj", "amazon", "pricing", "revenue"})) router_summary = ( "Router matched tokens: " + ", ".join(summary_tokens) if summary_tokens else "Router balanced coverage across experts" ) notes = " | ".join(router_notes) plan = MixtureOfExpertsPlan( router_summary=router_summary, maturity=maturity, activations=activations, evaluation_metrics=sorted(metrics) if metrics else ["roi", "ctr"], notes=notes, source="moe_router", ) return [plan] class AgenticLoopProcess: """Transforms prompts into continuous agentic loop blueprints.""" def __init__( self, loops: Sequence[Dict[str, Any]] | None = None, dataset_path: str = AGENTIC_LOOPS_DATASET, ) -> None: self._raw_loops = list(loops) if loops is not None else self._load(dataset_path) @staticmethod def _load(path: str) -> List[Dict[str, Any]]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] if isinstance(payload, list): return [entry for entry in payload if isinstance(entry, dict)] return [] @staticmethod def _tokenize(text: str) -> set[str]: return {token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2} def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 1, ) -> List[AgenticLoopPlan]: if not campaigns: raise ValueError("Agentic loop plans require campaign scores") if not self._raw_loops: return [] prompt_tokens = self._tokenize(prompt) avg_confidence = mean(score.metrics.confidence for score in campaigns) plans: List[AgenticLoopPlan] = [] for loop in self._raw_loops[: max(1, limit)]: trigger_keywords = {kw.lower() for kw in loop.get("trigger_keywords", [])} matched = trigger_keywords & prompt_tokens if avg_confidence >= 0.75 and matched: maturity = "autonomous" elif avg_confidence >= 0.65: maturity = "advanced" else: maturity = "foundational" reasoning_parts = [] if matched: reasoning_parts.append( f"Prompt references {', '.join(sorted(matched))} indicating readiness for agentic loop activation." ) reasoning_parts.append( f"Campaign confidence {avg_confidence:.0%} supports continuous execution without drift." ) reasoning = " ".join(reasoning_parts) stages: List[AgenticLoopStage] = [] for stage in loop.get("stages", []): stages.append( AgenticLoopStage( stage=stage.get("stage", "stage"), objective=stage.get("objective", ""), actions=list(stage.get("actions", [])), tools=list(stage.get("tools", [])), signals=list(stage.get("signals", [])), outputs=list(stage.get("outputs", [])), metrics=list(stage.get("metrics", [])), ) ) plans.append( AgenticLoopPlan( name=loop.get("name", "Agentic Loop"), goal_focus=loop.get("goal_focus", "marketing"), summary=loop.get("summary", ""), maturity=maturity, reasoning=reasoning, stages=stages, guardrails=list(loop.get("guardrails", [])), source=loop.get("source", "agentic_loop_dataset"), ) ) return plans class MiniQuantumSimulator: """Minimal single-qubit simulator used for deterministic stress scoring. The simulator now supports optional Aer/Cirq style execution when the dependencies are available, allows callers to control shot counts, and can perform lightweight Monte Carlo sampling when deterministic gates are not sufficient for stress harnesses. All fallbacks remain deterministic so the wider marketing ANI continues to operate without external services. """ def __init__(self, shots: int = 1024, backend: str | None = None) -> None: self._sqrt_half = 1 / math.sqrt(2) self._shots = max(1, shots) self._last_shots = self._shots self._rand = random.Random(1337) self._qiskit_available = False self._backend = None self._QuantumCircuit = None backend_name = backend or os.getenv("QUANTUM_SIM_BACKEND", "aer_simulator") try: # pragma: no cover - optional dependency wiring from qiskit import QuantumCircuit # type: ignore from qiskit_aer import Aer # type: ignore self._QuantumCircuit = QuantumCircuit try: self._backend = Aer.get_backend(backend_name) except Exception: # pragma: no cover - backend lookup self._backend = Aer.get_backend("aer_simulator") self._qiskit_available = True except Exception: # pragma: no cover - library optional self._qiskit_available = False @staticmethod def _parse_angle(text: str) -> float: value = text.strip().lower() if not value: return 0.0 multiplier = -1.0 if value.startswith("-") else 1.0 if value.startswith("-"): value = value[1:] if value == "pi": return multiplier * math.pi if value.startswith("pi/"): try: return multiplier * math.pi / float(value[3:]) except (TypeError, ValueError): return 0.0 if "*" in value and "pi" in value: left, right = value.split("*", 1) try: factor = float(left) except ValueError: factor = 1.0 if right == "pi": return multiplier * factor * math.pi if right.startswith("pi/"): try: return multiplier * factor * math.pi / float(right[3:]) except (TypeError, ValueError): return 0.0 try: return multiplier * float(value) except ValueError: return 0.0 def _matrix_for_gate(self, spec: str) -> Tuple[Tuple[complex, complex], Tuple[complex, complex]]: gate = spec.strip().upper() identity = ((1 + 0j, 0 + 0j), (0 + 0j, 1 + 0j)) if not gate: return identity if gate == "H": s = self._sqrt_half return ((s, s), (s, -s)) if gate == "X": return ((0 + 0j, 1 + 0j), (1 + 0j, 0 + 0j)) if gate == "Y": return ((0 + 0j, -1j), (1j, 0 + 0j)) if gate == "Z": return ((1 + 0j, 0 + 0j), (0 + 0j, -1 + 0j)) if gate.startswith("RX(") and gate.endswith(")"): angle = self._parse_angle(gate[3:-1]) c = math.cos(angle / 2.0) s = math.sin(angle / 2.0) return ((c, -1j * s), (-1j * s, c)) if gate.startswith("RY(") and gate.endswith(")"): angle = self._parse_angle(gate[3:-1]) c = math.cos(angle / 2.0) s = math.sin(angle / 2.0) return ((c, -s), (s, c)) if gate.startswith("RZ(") and gate.endswith(")"): angle = self._parse_angle(gate[3:-1]) half = angle / 2.0 return ( (complex(math.cos(-half), math.sin(-half)), 0 + 0j), (0 + 0j, complex(math.cos(half), math.sin(half))), ) if gate.startswith("PHASE(") and gate.endswith(")"): angle = self._parse_angle(gate[6:-1]) return ((1 + 0j, 0 + 0j), (0 + 0j, complex(math.cos(angle), math.sin(angle)))) return identity def _run_qiskit(self, gate_sequence: Sequence[str], shots: int) -> float: if not self._qiskit_available or not self._QuantumCircuit or not self._backend: return -1.0 circuit = self._QuantumCircuit(1, 1) for gate in gate_sequence: upper = gate.strip().upper() if not upper: continue if upper == "H": circuit.h(0) elif upper == "X": circuit.x(0) elif upper == "Y": circuit.y(0) elif upper == "Z": circuit.z(0) elif upper.startswith("RX(") and upper.endswith(")"): circuit.rx(self._parse_angle(upper[3:-1]), 0) elif upper.startswith("RY(") and upper.endswith(")"): circuit.ry(self._parse_angle(upper[3:-1]), 0) elif upper.startswith("RZ(") and upper.endswith(")"): circuit.rz(self._parse_angle(upper[3:-1]), 0) elif upper.startswith("PHASE(") and upper.endswith(")"): circuit.p(self._parse_angle(upper[6:-1]), 0) # fall back to no-op for unknown gates to keep compatibility circuit.measure(0, 0) try: # pragma: no cover - optional dependency from qiskit import execute # type: ignore job = execute(circuit, backend=self._backend, shots=shots) result = job.result() counts = result.get_counts() except Exception: # pragma: no cover - runtime fallback return -1.0 ones = counts.get("1", 0) return float(max(0.0, min(1.0, ones / shots))) def run( self, gate_sequence: Sequence[str], shots: int | None = None, monte_carlo: bool = False, ) -> float: active_shots = max(1, shots or self._shots) if self._qiskit_available: probability = self._run_qiskit(gate_sequence, active_shots) if probability >= 0.0: self._last_shots = active_shots return probability state = (1 + 0j, 0 + 0j) for gate in gate_sequence: matrix = self._matrix_for_gate(gate) state = ( matrix[0][0] * state[0] + matrix[0][1] * state[1], matrix[1][0] * state[0] + matrix[1][1] * state[1], ) probability_one = abs(state[1]) ** 2 if monte_carlo: successes = 0 for _ in range(active_shots): if self._rand.random() < probability_one: successes += 1 probability_one = successes / active_shots self._last_shots = active_shots return float(max(0.0, min(1.0, probability_one))) @property def available(self) -> bool: # pragma: no cover - simple property return True @property def shots(self) -> int: return self._shots @property def last_shots(self) -> int: return self._last_shots class QuantumSimulationProcess: """Generates quantum stress testing plans spanning revenue systems.""" REQUIRED_DOMAINS: set[str] = { "google_ads", "aws", "marketing", "bitcoin", "stockmarket", "stripe", "cashapp", "venmo", "paypal", "agentic", } ENV_HINTS: Dict[str, List[str]] = { "qiskit": ["QISKIT_API_TOKEN", "QISKIT_RUNTIME_API_KEY"], "aws_braket": ["AWS_BRAKET_REGION", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"], "google_cirq": ["GOOGLE_CIRQ_PROJECT", "GOOGLE_APPLICATION_CREDENTIALS"], "azure_quantum": ["AZURE_QUANTUM_WORKSPACE", "AZURE_QUANTUM_RESOURCE_ID"], } DOMAIN_ACTION_HINTS: Dict[str, str] = { "google_ads": "Sync quantum budget insights with Google Ads enrichment to guard ROAS.", "aws": "Blend simulator outputs with AWS revenue playbooks for quota planning.", "marketing": "Route funnel drift alerts into automation sprints and learning loops.", "bitcoin": "Pair quantum volatility scores with crypto market fallbacks.", "stockmarket": "Feed equity whipsaw metrics into scenario planning envelopes.", "stripe": "Stage redundant checkouts when Stripe latency probability climbs.", "cashapp": "Coordinate payout pacing with Cash App Pay readiness guides.", "venmo": "Tune social commerce messaging using Venmo stress telemetry.", "paypal": "Adjust PayPal reserve buffers based on quantum payout drag signals.", "agentic": "Throttle agent workloads when loop saturation probabilities spike.", } def __init__( self, capabilities: Sequence[QuantumSimulatorCapability] | None = None, dataset_path: str = QUANTUM_SIMULATIONS_DATASET, simulator: MiniQuantumSimulator | None = None, config: Dict[str, List[str]] | None = None, ) -> None: self._capabilities = ( list(capabilities) if capabilities is not None else self._load(dataset_path) ) self._simulator = simulator or MiniQuantumSimulator() self._config = config or self._config_from_env() @classmethod def _config_from_env(cls) -> Dict[str, List[str]]: config: Dict[str, List[str]] = {} for name, env_vars in cls.ENV_HINTS.items(): available = [env for env in env_vars if os.getenv(env)] if available: config[name] = sorted({env.lower() for env in available}) return config @property def configured(self) -> bool: return bool(self._config) @staticmethod def _tokenize(text: str) -> set[str]: return {token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2} @classmethod def _load(cls, path: str) -> List[QuantumSimulatorCapability]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except (FileNotFoundError, json.JSONDecodeError): payload = [] capabilities: List[QuantumSimulatorCapability] = [] if isinstance(payload, list): for entry in payload: if not isinstance(entry, dict): continue scenarios: List[QuantumScenarioTemplate] = [] for raw in entry.get("scenarios", []): if not isinstance(raw, dict): continue try: scenarios.append( QuantumScenarioTemplate( name=raw["name"], domain=raw["domain"], objective=raw["objective"], gate_sequence=list(raw.get("gate_sequence", [])), target_probability=float(raw.get("target_probability", 0.5)), stressors=list(raw.get("stressors", [])), metrics=list(raw.get("metrics", [])), integration_points=list(raw.get("integration_points", [])), playbook=list(raw.get("playbook", [])), notes=raw.get("notes"), ) ) except (KeyError, TypeError, ValueError): continue if not scenarios: continue capabilities.append( QuantumSimulatorCapability( simulator=entry.get("simulator", "Quantum Simulator"), provider=entry.get("provider", "open_source"), description=entry.get( "description", "Quantum-inspired simulator for revenue stress testing.", ), training_focus=list(entry.get("training_focus", [])), evaluation_metrics=list(entry.get("evaluation_metrics", [])), orchestration=list(entry.get("orchestration", [])), dependencies=list(entry.get("dependencies", [])), docs=list(entry.get("docs", [])), scenarios=scenarios, source=entry.get("source", "quantum_simulation_catalog"), notes=entry.get("notes"), ) ) if not capabilities: fallback_scenarios: List[QuantumScenarioTemplate] = [] for domain in sorted(cls.REQUIRED_DOMAINS): fallback_scenarios.append( QuantumScenarioTemplate( name=f"{domain.title()} Stress Harness", domain=domain, objective=f"Baseline quantum stress for {domain.replace('_', ' ')} stacks.", gate_sequence=["H", "RX(pi/4)", "RZ(pi/6)", "H"], target_probability=0.6, stressors=["throughput", "latency"], metrics=["stability_index"], integration_points=[], playbook=["Run fallback simulator to validate coverage."], notes="Fallback coverage generated when dataset missing.", ) ) capabilities.append( QuantumSimulatorCapability( simulator="Fallback Quantum Simulator", provider="deterministic", description="Deterministic single-qubit simulator for baseline stress coverage.", training_focus=["Quantum-inspired stress heuristics"], evaluation_metrics=["stability_index"], orchestration=["Upgrade dataset to unlock provider-specific playbooks."], dependencies=[], docs=[], scenarios=fallback_scenarios, source="quantum_simulation_fallback", notes="Generated offline fallback to preserve ANI readiness.", ) ) return capabilities def plans( self, prompt: str, campaigns: Sequence["CampaignScore"], limit: int = 1, ) -> List[QuantumStressTestPlan]: if not campaigns: raise ValueError("Quantum simulator plans require campaign scores") prompt_tokens = self._tokenize(prompt) campaign_tokens = self._tokenize( " ".join( " ".join( [ score.campaign.channel, score.campaign.audience, score.campaign.offer, score.campaign.objective, ] ) for score in campaigns ) ) plans: List[Tuple[float, QuantumStressTestPlan]] = [] for capability in self._capabilities: stress_tests: List[QuantumStressTest] = [] coverage: set[str] = set() stability_scores: List[float] = [] per_test_agents: List[int] = [] per_test_nodes: List[int] = [] per_test_compression: List[float] = [] per_test_shots: List[int] = [] domain_hits = 0 domain_stability_map: Dict[str, float] = {} for template in capability.scenarios: coverage.add(template.domain.lower()) measurement = self._simulator.run(template.gate_sequence) stability = 1.0 - abs(measurement - template.target_probability) stability = max(0.0, min(1.0, stability)) stability_scores.append(stability) domain_stability_map[template.domain.lower()] = round(stability, 4) shots_used = getattr( self._simulator, "last_shots", getattr(self._simulator, "shots", 1024), ) stability_floor = max(0.05, stability) agents_per_node = max(1, int(round(1_000_000 * stability_floor))) nodes_for_one_million = max(1, math.ceil(1_000_000 / agents_per_node)) compression_ratio = round( min(1.0, stability_floor / max(1, len(template.gate_sequence)) * 1.5), 6, ) per_test_agents.append(agents_per_node) per_test_nodes.append(nodes_for_one_million) per_test_compression.append(compression_ratio) per_test_shots.append(int(shots_used)) domain_tokens = self._tokenize(template.domain.replace("_", " ")) if domain_tokens & prompt_tokens or domain_tokens & campaign_tokens: domain_hits += 1 recommended_actions = list(template.playbook) hint = self.DOMAIN_ACTION_HINTS.get(template.domain.lower()) if hint and hint not in recommended_actions: recommended_actions.append(hint) if template.domain.lower() == "bitcoin": recommended_actions.append( "Blend quantum liquidity stress with crypto telemetry to hedge exposure." ) stress_tests.append( QuantumStressTest( name=template.name, domain=template.domain, objective=template.objective, stressors=list(template.stressors), metrics=list(template.metrics), gate_sequence=list(template.gate_sequence), measurement_probability=round(measurement, 4), stability_index=round(stability, 4), integration_points=list(template.integration_points), recommended_actions=recommended_actions, notes=template.notes, source=capability.source, shots_used=int(shots_used), agents_per_node=agents_per_node, nodes_for_one_million=nodes_for_one_million, compression_ratio=compression_ratio, ) ) if not stress_tests: continue coverage_ratio = len(coverage & self.REQUIRED_DOMAINS) / len(self.REQUIRED_DOMAINS) avg_stability = mean(stability_scores) score = 0.45 + coverage_ratio * 0.4 + avg_stability * 0.3 + 0.05 * domain_hits if self.configured: score += 0.15 if {"quantum", "simulation", "stress"} & prompt_tokens: score += 0.08 if self.configured and coverage_ratio >= 0.8 and avg_stability >= 0.55: maturity = "autonomous" elif coverage_ratio >= 0.6 or avg_stability >= 0.5: maturity = "advanced" else: maturity = "foundational" domains_sorted = sorted(coverage & self.REQUIRED_DOMAINS) reasoning_parts = [] if domains_sorted: reasoning_parts.append( f"Simulator covers {len(domains_sorted)} critical domains ({', '.join(domains_sorted[:5])})." ) reasoning_parts.append(f"Average stability index {avg_stability:.2f} vs targets {mean(t.target_probability for t in capability.scenarios):.2f}.") if self.configured: providers = ", ".join(sorted(self._config.keys())) reasoning_parts.append(f"Live credentials detected for {providers} environments.") reasoning = " ".join(reasoning_parts) orchestration_steps = list(capability.orchestration) orchestration_steps.append( "Distribute quantum telemetry to statistical reasoning, scenario planning, and excellence scoring." ) orchestration_steps.append( "Re-run stress harness after major campaign, budget, or payment changes." ) dependencies = list(capability.dependencies) if not self.configured: dependencies.append( "Configure Qiskit/AWS Braket/Google Cirq credentials to graduate to live simulations." ) source = capability.source if self.configured: source = "quantum_simulation_live" notes_parts: List[str] = [] if capability.notes: notes_parts.append(capability.notes) if coverage_ratio < 1.0: missing = sorted(self.REQUIRED_DOMAINS - coverage) if missing: notes_parts.append( "Additional coverage recommended for " + ", ".join(missing[:3]) + ("..." if len(missing) > 3 else "") ) if avg_stability < 0.45: notes_parts.append("Consider increasing simulator shots or refining gate depth for stability.") plan_agents_per_node = min(per_test_agents) if per_test_agents else 1 plan_nodes_for_million = max(per_test_nodes) if per_test_nodes else 1 plan_compression_index = ( round(mean(per_test_compression), 4) if per_test_compression else 0.0 ) plan_shots = max(per_test_shots) if per_test_shots else getattr( self._simulator, "shots", 1024 ) continuous_operation = all(score >= 0.5 for score in stability_scores) plan = QuantumStressTestPlan( simulator=capability.simulator, provider=capability.provider, maturity=maturity, description=capability.description, reasoning=reasoning, training_focus=list(capability.training_focus), stress_tests=stress_tests, orchestration_steps=orchestration_steps, metrics=list(capability.evaluation_metrics or ["stability_index"]), dependencies=dependencies, docs=list(capability.docs), coverage_ratio=round(coverage_ratio, 4), domain_stability=domain_stability_map, agents_per_node=plan_agents_per_node, nodes_for_one_million=plan_nodes_for_million, compression_index=plan_compression_index, shots=int(plan_shots), continuous_operation=continuous_operation, source=source, notes=" | ".join(notes_parts) if notes_parts else None, ) plans.append((score, plan)) plans.sort(key=lambda item: item[0], reverse=True) return [plan for _, plan in plans[: max(1, limit)]] @dataclass(slots=True) class AWSRevenueProgram: """AWS program metadata for revenue acceleration.""" program_name: str focus: str description: str revenue_motion: str primary_service: str secondary_services: List[str] automation_play: str base_roi: float focus_keywords: List[str] @dataclass(slots=True) class AWSRevenueRecommendation: """Recommendation payload produced by the AWS revenue process.""" program_name: str focus: str description: str revenue_motion: str primary_service: str secondary_services: List[str] automation_play: str projected_roi: float source: str notes: str | None = None class AWSGrowthPlaybookCatalog: """Loads AWS growth programs and matches them to prompts.""" def __init__(self, programs: Sequence[AWSRevenueProgram] | None = None): if programs is not None: self._programs = list(programs) else: with open(AWS_GROWTH_PLAYBOOKS, "r", encoding="utf-8") as handle: data = json.load(handle) self._programs = [AWSRevenueProgram(**entry) for entry in data] def match_programs( self, tokens: Sequence[str], limit: int ) -> List[Tuple[float, AWSRevenueProgram]]: ranked: List[Tuple[float, AWSRevenueProgram]] = [] token_set = {token.lower() for token in tokens} for program in self._programs: haystack = ( f"{program.program_name} {program.description} {program.revenue_motion} " f"{program.primary_service} {' '.join(program.secondary_services)}" ).lower() score = 0.0 for token in token_set: if token in haystack: score += 1.2 if token in {kw.lower() for kw in program.focus_keywords}: score += 1.6 if program.focus in token_set: score += 0.5 ranked.append((score, program)) ranked.sort(key=lambda item: item[0], reverse=True) return ranked[: max(1, limit)] class AWSRevenueProcess: """Generates AWS-backed revenue motions and optionally validates via boto3.""" REQUIRED_ENV_VARS = { "AWS_ACCESS_KEY_ID": "access_key_id", "AWS_SECRET_ACCESS_KEY": "[credential removed]", } OPTIONAL_ENV_VARS = { "AWS_SESSION_TOKEN": "session_token", "AWS_REGION": "region", "AWS_PROFILE": "profile", } def __init__( self, catalog: AWSGrowthPlaybookCatalog | None = None, config: Dict[str, Any] | None = None, ) -> None: self._catalog = catalog or AWSGrowthPlaybookCatalog() self._config = config or self._config_from_env() self._session = self._build_session() def _build_session(self) -> Any | None: if not self.configured or boto3 is None: return None try: # pragma: no cover - requires boto3 dependency return boto3.session.Session( aws_access_key_id=self._config.get("access_key_id"), aws_secret_access_key=[credential removed]"secret_access_key"), aws_session_token=self._config.get("session_token"), region_name=self._config.get("region"), profile_name=self._config.get("profile"), ) except Exception: return None @classmethod def _config_from_env(cls) -> Dict[str, Any]: config: Dict[str, Any] = {} for env_var, key in {**cls.REQUIRED_ENV_VARS, **cls.OPTIONAL_ENV_VARS}.items(): value = os.getenv(env_var) if value: config[key] = value return config @property def configured(self) -> bool: required = {key for key in self.REQUIRED_ENV_VARS.values()} return required <= self._config.keys() def plays( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int, ) -> List[AWSRevenueRecommendation]: tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower()) matches = self._catalog.match_programs(tokens, limit) if not matches: return [] avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.5 avg_conf = mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.65 recommendations: List[AWSRevenueRecommendation] = [] for match_score, program in matches: projected_roi = round( program.base_roi + match_score * 0.2 + max(0.0, avg_roi - 2.0) * 0.15, 2, ) projected_roi = max(projected_roi, program.base_roi) rationale = ( f"Leverage {program.primary_service} with {', '.join(program.secondary_services)} " f"to execute {program.revenue_motion.lower()}" ) notes = None if not self._session: notes = ( "Offline AWS growth playbooks active; configure AWS credentials to validate deployment." ) recommendations.append( AWSRevenueRecommendation( program_name=program.program_name, focus=program.focus, description=program.description, revenue_motion=program.revenue_motion, primary_service=program.primary_service, secondary_services=program.secondary_services, automation_play=program.automation_play, projected_roi=max(projected_roi, avg_conf * program.base_roi), source="aws_sdk" if self._session else "aws_playbooks", notes=notes, ) ) return recommendations class BudgetAllocatorProcess: """Turns campaign rankings into channel-level budget plans.""" def __init__( self, profiles: Sequence[BudgetProfile] | None = None, dataset_path: str = CHANNEL_INVESTMENTS_DATASET, default_budget: float = 25000.0, ): self._profiles = list(profiles) if profiles is not None else self._profiles_from_file(dataset_path) self._default_budget = default_budget @staticmethod def _profiles_from_file(path: str) -> List[BudgetProfile]: try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except FileNotFoundError: payload = {"profiles": []} profiles: List[BudgetProfile] = [] for entry in payload.get("profiles", []): profiles.append( BudgetProfile( channel=entry.get("channel", "Integrated Marketing"), keywords=[keyword.lower() for keyword in entry.get("keywords", [])], baseline_cpc=float(entry.get("baseline_cpc", 4.0)), baseline_cpl=float(entry.get("baseline_cpl", 150.0)), avg_conversion_rate=float(entry.get("avg_conversion_rate", 0.045)), avg_roi=float(entry.get("avg_roi", 3.2)), recommended_monthly_spend=float(entry.get("recommended_monthly_spend", 20000.0)), ) ) if not profiles: profiles.append( BudgetProfile( channel="Integrated Marketing", keywords=["integrated", "multi-channel"], baseline_cpc=4.0, baseline_cpl=150.0, avg_conversion_rate=0.045, avg_roi=3.0, recommended_monthly_spend=20000.0, ) ) return profiles def _match_profile(self, channel: str) -> BudgetProfile: lowered = channel.lower() best: BudgetProfile | None = None best_score = -1 for profile in self._profiles: score = 0 if profile.channel.lower() in lowered: score += 3 for keyword in profile.keywords: if keyword in lowered: score += 1 if score > best_score: best = profile best_score = score if best is None: best = BudgetProfile( channel=channel, keywords=[channel.lower()], baseline_cpc=4.5, baseline_cpl=165.0, avg_conversion_rate=0.042, avg_roi=3.1, recommended_monthly_spend=self._default_budget, ) return best @staticmethod def _parse_budget_from_prompt(prompt: str) -> float | None: pattern = re.compile(r"\$?([0-9][0-9,]*)(?:\s*(k|m|b|million|billion|thousand))?", re.IGNORECASE) matches = pattern.findall(prompt) if not matches: return None inferred: List[float] = [] for value_str, multiplier in matches: value = float(value_str.replace(",", "")) if multiplier: token = multiplier.lower() if token in {"k", "thousand"}: value *= 1_000 elif token in {"m", "million"}: value *= 1_000_000 elif token in {"b", "billion"}: value *= 1_000_000_000 elif value < 100: value *= 1_000 inferred.append(value) return max(inferred) if inferred else None def _infer_budget(self, prompt: str, profiles: Sequence[BudgetProfile]) -> float: parsed = self._parse_budget_from_prompt(prompt) if parsed is not None and parsed > 0: return float(parsed) if profiles: average_spend = mean(profile.recommended_monthly_spend for profile in profiles) return float(max(average_spend, self._default_budget)) return self._default_budget @staticmethod def _allocation_weight(score: CampaignScore, profile: BudgetProfile) -> float: roi_component = score.metrics.expected_roi * 0.45 ctr_component = score.metrics.expected_ctr * 100.0 * 0.2 profile_component = profile.avg_roi * 0.25 efficiency_component = 240.0 / max(profile.baseline_cpl, 60.0) return max(0.1, roi_component + ctr_component + profile_component + efficiency_component) @staticmethod def _summarize_plan(budget: float, expected_revenue: float, count: int) -> str: if budget <= 0 or expected_revenue <= 0: return "No budgetable campaigns identified." roi = expected_revenue / budget if budget else 0 return ( f"Deploy ${budget:,.0f} across {count} channels targeting ~${expected_revenue:,.0f} revenue " f"(~{roi:.1f}x ROI)." ) @staticmethod def _payback_weeks(budget: float, expected_revenue: float) -> float: profit = expected_revenue - budget if budget <= 0: return 0.0 if profit <= 0: return 16.0 weeks = (budget / profit) * 4 return round(max(2.0, min(26.0, weeks)), 1) def plan( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int = 3, ) -> BudgetPlan: if not campaigns: raise ValueError("plan requires at least one campaign score") selected = list(campaigns[: max(1, limit)]) profiles = [self._match_profile(score.campaign.channel) for score in selected] total_budget = self._infer_budget(prompt, profiles) weighted: List[Tuple[CampaignScore, BudgetProfile, float]] = [] for score, profile in zip(selected, profiles): weight = self._allocation_weight(score, profile) weighted.append((score, profile, weight)) total_weight = sum(entry[2] for entry in weighted) allocations: List[BudgetAllocation] = [] running_total = 0.0 for index, (score, profile, weight) in enumerate(weighted): share = weight / total_weight if total_weight else 1.0 / len(weighted) allocation = round(total_budget * share, 2) if index == len(weighted) - 1: allocation = round(total_budget - running_total, 2) running_total += allocation expected_leads = round(allocation / max(profile.baseline_cpl, 1.0), 1) blended_roi = max(1.05, (score.metrics.expected_roi + profile.avg_roi) / 2) expected_revenue = round(allocation * blended_roi, 2) rationale = ( f"Amplify {score.campaign.channel} for {score.campaign.audience} to accelerate " f"{score.campaign.objective}." ) notes = None if profile.channel.lower() not in score.campaign.channel.lower(): notes = ( f"Mapped to {profile.channel} benchmarks using keyword similarity; override dataset " "to tune allocation." ) allocations.append( BudgetAllocation( channel=score.campaign.channel, audience_focus=score.campaign.audience, offer=score.campaign.offer, objective=score.campaign.objective, allocation_usd=allocation, allocation_weight=round(share, 3), expected_leads=expected_leads, expected_revenue=expected_revenue, rationale=rationale, source="budget_optimizer", notes=notes, ) ) total_expected_revenue = round(sum(entry.expected_revenue for entry in allocations), 2) expected_roi = round(total_expected_revenue / total_budget, 2) if total_budget else 0.0 payback_weeks = self._payback_weeks(total_budget, total_expected_revenue) summary = self._summarize_plan(total_budget, total_expected_revenue, len(allocations)) return BudgetPlan( total_budget=round(total_budget, 2), allocations=allocations, expected_revenue=total_expected_revenue, expected_roi=expected_roi, payback_period_weeks=payback_weeks, summary=summary, ) @dataclass(slots=True) class GoogleCampaignPerformance: """Performance snapshot surfaced from the Google Ads API.""" campaign_name: str clicks: int impressions: int conversions: float cost_micros: int ctr: float cpc: float roas: float source: str notes: str | None = None @dataclass(slots=True) class StripeMonetizationPlan: """Stripe checkout strategy synthesized from campaign intelligence.""" product_name: str price_usd: float billing_interval: str currency: str rationale: str setup_steps: List[str] source: str notes: str | None = None class StripeProcess: """Shapes campaign ideas into Stripe-ready monetization workflows.""" def __init__(self, api_key: str | None = None, account_id: str | None = None) -> None: self._api_key = [credential removed] or os.getenv("STRIPE_API_KEY") self._account_id = account_id or os.getenv("STRIPE_ACCOUNT_ID") self._configured = bool(stripe is not None and self._api_key) if self._configured and stripe is not None: # pragma: no cover - requires SDK stripe.api_key = [credential removed] @property def configured(self) -> bool: return self._configured def _infer_billing_interval(self, prompt: str) -> str: lowered = prompt.lower() subscription_keywords = { "subscription", "recurring", "membership", "retainer", "saas", } if any(keyword in lowered for keyword in subscription_keywords): return "monthly" if "annual" in lowered or "yearly" in lowered: return "annual" return "one_time" def plans( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int = 3, ) -> List[StripeMonetizationPlan]: if not campaigns: raise ValueError("Stripe plans require at least one campaign score") interval = self._infer_billing_interval(prompt) plans: List[StripeMonetizationPlan] = [] for scored in campaigns[: max(1, limit)]: base_price = max(29.0, scored.metrics.expected_roi * 45) audience = scored.campaign.audience.lower() if "enterprise" in audience: base_price *= 1.6 elif "consumer" in audience: base_price *= 0.75 if interval == "annual": base_price *= 10.5 elif interval == "one_time": base_price *= 1.25 price = round(base_price, 2) rationale = ( f"Stripe-ready {interval.replace('_', ' ')} offer leveraging {scored.campaign.channel} " f"for {scored.campaign.audience}." ) if scored.metrics.confidence < 0.5: rationale += " Calibrate pricing with phased rollout given confidence score." setup_steps = [ "Create Product in Stripe aligned to the campaign offer.", "Attach a Price with the recommended interval and USD currency.", "Launch via Stripe Checkout or Payment Links to validate demand.", ] if interval != "one_time": setup_steps.append("Enable subscription trials and dunning automations.") else: setup_steps.append("Automate follow-up journeys via Stripe Webhooks or customer portal.") source = "stripe_sdk" if self._configured else "simulated" notes = None if not self._configured: notes = ( "Install the official stripe SDK and export STRIPE_API_KEY/STRIPE_ACCOUNT_ID " "to activate live checkout orchestration." ) plans.append( StripeMonetizationPlan( product_name=f"{scored.campaign.offer.title()} Checkout", price_usd=price, billing_interval=interval, currency="usd", rationale=rationale, setup_steps=setup_steps, source=source, notes=notes, ) ) return plans @dataclass(slots=True) class PayPalMonetizationPlan: """PayPal commerce strategy to operationalize campaign monetization.""" product_name: str price_usd: float checkout_flow: str payout_schedule: str rationale: str setup_steps: List[str] source: str notes: str | None = None class PayPalProcess: """Designs PayPal-ready offers so campaigns translate into paychecks.""" def __init__( self, client_id: str | None = None, client_secret: str | None = None, merchant_id: str | None = None, environment: str | None = None, ) -> None: self._client_id = client_id or os.getenv("PAYPAL_CLIENT_ID") self._client_secret = [credential removed] or os.getenv("PAYPAL_CLIENT_SECRET") self._merchant_id = merchant_id or os.getenv("PAYPAL_MERCHANT_ID") self._environment = (environment or os.getenv("PAYPAL_ENVIRONMENT", "sandbox")).lower() self._configured = bool( PayPalHttpClient is not None and self._client_id and self._client_secret ) self._client: PayPalHttpClient | None = None if self._configured and PayPalHttpClient is not None: # pragma: no cover - requires SDK if self._environment in {"live", "production"} and LiveEnvironment is not None: env = LiveEnvironment(self._client_id, self._client_secret) else: env = SandboxEnvironment(self._client_id, self._client_secret) self._client = PayPalHttpClient(env) @property def configured(self) -> bool: return self._configured def _infer_checkout_flow(self, prompt: str) -> str: lowered = prompt.lower() if any(keyword in lowered for keyword in ["subscription", "membership", "recurring"]): return "subscription" if any(keyword in lowered for keyword in ["invoice", "retainer", "contract"]): return "invoice" return "one_time" def _infer_payout_schedule(self, prompt: str) -> str: lowered = prompt.lower() if any(keyword in lowered for keyword in ["daily", "paycheck", "cash flow"]): return "daily" if any(keyword in lowered for keyword in ["weekly", "bi-weekly", "payout"]): return "weekly" return "rolling" def plans( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int = 3, ) -> List[PayPalMonetizationPlan]: if not campaigns: raise ValueError("PayPal plans require at least one campaign score") flow = self._infer_checkout_flow(prompt) payout_schedule = self._infer_payout_schedule(prompt) plans: List[PayPalMonetizationPlan] = [] for scored in campaigns[: max(1, limit)]: base_price = max(19.0, scored.metrics.expected_roi * 35) if "enterprise" in scored.campaign.audience.lower(): base_price *= 1.4 elif "consumer" in scored.campaign.audience.lower(): base_price *= 0.8 if flow == "subscription": base_price *= 9.5 elif flow == "invoice": base_price *= 4.5 price = round(base_price, 2) rationale = ( f"PayPal {flow.replace('_', ' ')} monetization aligned to {scored.campaign.channel} " f"for {scored.campaign.audience}." ) if payout_schedule == "daily": rationale += " Configured for accelerated daily paychecks." setup_steps = [ "Create Product or Plan in PayPal matching the campaign offer.", f"Configure {flow} checkout with contextual upsells.", f"Enable payouts on a {payout_schedule} cadence for predictable cashflow.", ] if self._merchant_id: setup_steps.append( f"Link merchant account {self._merchant_id} to PayPal Payouts for revenue capture." ) else: setup_steps.append("Map payouts to finance workflows for accounting reconciliation.") source = "paypal_sdk" if self._configured else "simulated" notes = None if not self._configured: notes = ( "Install paypalcheckoutsdk and export PAYPAL_CLIENT_ID/PAYPAL_CLIENT_SECRET " "to activate live PayPal revenue collection." ) plans.append( PayPalMonetizationPlan( product_name=f"{scored.campaign.offer.title()} PayPal Flow", price_usd=price, checkout_flow=flow, payout_schedule=payout_schedule, rationale=rationale, setup_steps=setup_steps, source=source, notes=notes, ) ) return plans class GoogleAdsProcess: """Bridges the Google Ads API to enrich ANI recommendations.""" REQUIRED_ENV_VARS = { "GOOGLE_ADS_DEVELOPER_TOKEN": "developer_token", "GOOGLE_ADS_CUSTOMER_ID": "customer_id", } OAUTH_ENV_VARS = { "GOOGLE_ADS_CLIENT_ID": "client_id", "GOOGLE_ADS_CLIENT_SECRET": "[credential removed]", "GOOGLE_ADS_REFRESH_TOKEN": "refresh_token", } OPTIONAL_ENV_VARS = { "GOOGLE_ADS_LOGIN_CUSTOMER_ID": "login_customer_id", } SERVICE_ACCOUNT_ENV_VARS = { "GOOGLE_ADS_SERVICE_ACCOUNT_INFO": "service_account_info", "GOOGLE_ADS_SERVICE_ACCOUNT_FILE": "service_account_file", "GOOGLE_ADS_SERVICE_ACCOUNT_B64": "service_account_b64", "GOOGLE_APPLICATION_CREDENTIALS": "service_account_file", } def __init__(self, config: Dict[str, Any] | None = None): self._config = config or self._config_from_env() self._client = self._build_client(self._config) @classmethod def _config_from_env(cls) -> Dict[str, Any]: config: Dict[str, Any] = {} for env_var, key in cls.REQUIRED_ENV_VARS.items(): value = os.getenv(env_var) if value: config[key] = value for env_var, key in {**cls.OAUTH_ENV_VARS, **cls.OPTIONAL_ENV_VARS}.items(): value = os.getenv(env_var) if value: config[key] = value for env_var, key in cls.SERVICE_ACCOUNT_ENV_VARS.items(): value = os.getenv(env_var) if not value: continue if key == "service_account_b64": try: decoded = base64.b64decode(value).decode("utf-8") config["service_account_info"] = json.loads(decoded) except (ValueError, json.JSONDecodeError): continue elif key == "service_account_file": try: with open(value, "r", encoding="utf-8") as handle: config["service_account_info"] = json.load(handle) except (FileNotFoundError, json.JSONDecodeError, OSError): continue elif key == "service_account_info": try: config["service_account_info"] = json.loads(value) except json.JSONDecodeError: continue return config @staticmethod def _build_client(config: Dict[str, Any]) -> GoogleAdsClient | None: if not config or "customer_id" not in config: return None if GoogleAdsClient is None: return None try: # pragma: no cover - relies on network credentials to execute credentials = None if config.get("service_account_info") and service_account is not None: info = config["service_account_info"] if isinstance(info, str): try: info = json.loads(info) except json.JSONDecodeError: info = None if isinstance(info, dict): scopes = ["https://www.googleapis.com/auth/adwords"] credentials = service_account.Credentials.from_service_account_info( info, scopes=scopes ) client_kwargs = { "developer_token": config.get("developer_token"), "login_customer_id": config.get("login_customer_id"), "use_proto_plus": True, } if credentials is not None: return GoogleAdsClient.load_from_dict(client_kwargs, credentials=credentials) # Fallback to OAuth credential set when present if all(config.get(field) for field in ("client_id", "client_secret", "refresh_token")): client_kwargs.update( { "client_id": config.get("client_id"), "client_secret": config.get("client_secret"), "refresh_token": config.get("refresh_token"), } ) return GoogleAdsClient.load_from_dict(client_kwargs) except Exception: # pragma: no cover - fallback handled downstream return None @property def configured(self) -> bool: return bool(self._client) def performance( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int, ) -> List[GoogleCampaignPerformance]: limit = max(1, limit) if self._client is None: return self._simulate(prompt, campaigns, limit) try: # pragma: no cover - executed when Google Ads credentials available ga_service = self._client.get_service("GoogleAdsService") query = self._build_query(limit) response = ga_service.search( customer_id=self._config.get("customer_id"), query=query, ) insights: List[GoogleCampaignPerformance] = [] for row in response: metrics = row.metrics campaign = row.campaign clicks = int(getattr(metrics, "clicks", 0)) impressions = int(getattr(metrics, "impressions", 0)) conversions = float(getattr(metrics, "conversions", 0.0)) cost_micros = int(getattr(metrics, "cost_micros", 0)) ctr = float(getattr(metrics, "ctr", 0.0)) cpc = float(getattr(metrics, "average_cpc", 0.0)) roas = float(getattr(metrics, "conversions_value_per_cost", 0.0)) insights.append( GoogleCampaignPerformance( campaign_name=getattr(campaign, "name", "Campaign"), clicks=clicks, impressions=impressions, conversions=conversions, cost_micros=cost_micros, ctr=ctr, cpc=cpc, roas=roas, source="google_ads_api", notes="Live Google Ads data", ) ) if insights: return insights except Exception: # pragma: no cover - gracefully fallback on API errors pass return self._simulate(prompt, campaigns, limit) @staticmethod def _build_query(limit: int) -> str: return ( "SELECT campaign.name, metrics.clicks, metrics.impressions, metrics.conversions, " "metrics.cost_micros, metrics.ctr, metrics.average_cpc, metrics.conversions_value_per_cost " "FROM campaign WHERE segments.date DURING LAST_30_DAYS " "ORDER BY metrics.conversions DESC " f"LIMIT {limit}" ) def _simulate( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int ) -> List[GoogleCampaignPerformance]: simulated: List[GoogleCampaignPerformance] = [] tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower()) emphasis = 1 + len(set(tokens)) * 0.02 for index, scored in enumerate(campaigns[:limit]): base_clicks = max(50, int(scored.score * 10)) clicks = int(base_clicks * emphasis) impressions = max(clicks * 8, 500) conversions = round(max(1.0, scored.metrics.expected_roi * 1.5), 2) cost_micros = int(max(1_000_000, impressions * 120)) ctr = round(min(0.25, clicks / max(1, impressions)), 4) cpc = round(cost_micros / max(1, clicks) / 1_000_000, 2) roas = round(scored.metrics.expected_roi / max(0.1, cpc), 2) simulated.append( GoogleCampaignPerformance( campaign_name=f"{scored.campaign.channel.title()} - Simulated", clicks=clicks, impressions=impressions, conversions=conversions, cost_micros=cost_micros, ctr=ctr, cpc=cpc, roas=roas, source="simulated", notes=( "Simulated from marketing benchmarks; set GOOGLE_ADS_* env vars " "for live Google Ads enrichment." ), ) ) if not simulated: simulated.append( GoogleCampaignPerformance( campaign_name="Simulated Campaign", clicks=120, impressions=2400, conversions=3.0, cost_micros=2500000, ctr=0.05, cpc=0.5, roas=4.2, source="simulated", notes="Fallback simulation due to missing campaigns", ) ) return simulated class CJAffiliateCatalog: """Loads CJ affiliate offers and surfaces relevant matches.""" def __init__(self, offers: Sequence[CJAffiliateOffer]): self._offers = list(offers) @classmethod def from_file(cls, path: str = CJ_OFFERS_DATASET) -> "CJAffiliateCatalog": try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except FileNotFoundError: return cls([]) except json.JSONDecodeError: return cls([]) offers: List[CJAffiliateOffer] = [] for entry in payload if isinstance(payload, list) else []: try: offer = CJAffiliateOffer( program_name=entry["program_name"], advertiser_id=str(entry["advertiser_id"]), category=entry.get("category", "Unknown"), commission_rate=float(entry.get("commission_rate", 0.0)), epc=float(entry.get("epc", 0.0)), seven_day_epc=float(entry.get("seven_day_epc", entry.get("epc", 0.0))), network_rank=int(entry.get("network_rank", 7)), payout_terms=entry.get("payout_terms", "Net30"), landing_url=entry.get("landing_url", ""), description=entry.get("description", ""), keywords=list(entry.get("keywords", [])), ) except (KeyError, TypeError, ValueError): continue offers.append(offer) return cls(offers) def match_offers( self, tokens: Sequence[str], limit: int = 3, focus: str | None = None, ) -> List[Tuple[float, CJAffiliateOffer]]: limit = max(1, limit) scored: List[Tuple[float, CJAffiliateOffer]] = [] token_set = {token.lower() for token in tokens if token} for offer in self._offers: keywords = {keyword.lower() for keyword in offer.keywords} description = offer.description.lower() category = offer.category.lower() score = offer.epc / 120.0 score += max(0.0, (6 - min(offer.network_rank, 6)) * 0.1) for token in token_set: if token in keywords: score += 1.2 elif token in description or token in category: score += 0.8 if focus and focus in category: score += 0.5 scored.append((score, offer)) scored.sort(key=lambda item: item[0], reverse=True) return scored[:limit] class CJAffiliateProcess: """Transforms CJ affiliate data into monetization playbooks.""" REQUIRED_ENV_VARS = { "CJ_DEVELOPER_KEY": "developer_key", } OPTIONAL_ENV_VARS = { "CJ_WEBSITE_ID": "website_id", "CJ_ACCOUNT_ID": "account_id", } def __init__( self, catalog: CJAffiliateCatalog | None = None, config: Dict[str, Any] | None = None, ): self._catalog = catalog or CJAffiliateCatalog.from_file() self._config = config or self._config_from_env() @classmethod def _config_from_env(cls) -> Dict[str, Any]: config: Dict[str, Any] = {} for env_var, key in {**cls.REQUIRED_ENV_VARS, **cls.OPTIONAL_ENV_VARS}.items(): value = os.getenv(env_var) if value: config[key] = value return config @property def configured(self) -> bool: return "developer_key" in self._config def offers( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int, ) -> List[CJAffiliateRecommendation]: tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower()) focus = None if campaigns: focus = campaigns[0].campaign.channel.lower().split()[0] matches = self._catalog.match_offers(tokens, limit, focus) if not matches: return [] avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.5 avg_ctr = mean([score.metrics.expected_ctr for score in campaigns]) if campaigns else 0.03 enriched: List[CJAffiliateRecommendation] = [] for match_score, offer in matches: keyword_matches = sorted({token for token in tokens if token in {k.lower() for k in offer.keywords}}) intensity = max(0.15, min(1.8, match_score / 4.0)) projected_epc = round(offer.epc * (1 + intensity * 0.18 + avg_ctr * 4), 2) projected_revenue = round(projected_epc * max(5.0, avg_roi * 6), 2) if keyword_matches: rationale = ( f"Matches {', '.join(keyword_matches)} demand with {offer.category.lower()} payouts." ) else: rationale = f"High network rank offer aligned to {offer.category.lower()} intent." source = "cj_network" if self.configured else "cj_catalog" notes = None if not self.configured: notes = ( "Simulated from CJ offer corpus; set CJ_DEVELOPER_KEY and related env vars " "to activate live API harvesting." ) enriched.append( CJAffiliateRecommendation( program_name=offer.program_name, advertiser_id=offer.advertiser_id, category=offer.category, commission_rate=round(offer.commission_rate, 4), epc=round(offer.epc, 2), seven_day_epc=round(offer.seven_day_epc, 2), projected_epc=projected_epc, projected_monthly_revenue=projected_revenue, network_rank=offer.network_rank, payout_terms=offer.payout_terms, landing_url=offer.landing_url, rationale=rationale, source=source, notes=notes, ) ) return enriched class InfluencerCollaborationProcess: """Matches prompts with influencer partners to accelerate revenue loops.""" OPTIONAL_ENV_VARS = { "INFLUENCER_NETWORK_API_KEY": "[credential removed]", "INFLUENCER_NETWORK_WORKSPACE": "workspace", } def __init__( self, catalog: InfluencerNetworkCatalog | None = None, config: Dict[str, Any] | None = None, ) -> None: self._catalog = catalog or InfluencerNetworkCatalog() self._config = config or self._config_from_env() @classmethod def _config_from_env(cls) -> Dict[str, Any]: config: Dict[str, Any] = {} for env_var, key in cls.OPTIONAL_ENV_VARS.items(): value = os.getenv(env_var) if value: config[key] = value return config @property def configured(self) -> bool: return "api_key" in self._config def recommendations( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int, ) -> List[InfluencerRecommendation]: tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower()) focus = None if campaigns: primary = campaigns[0].campaign focus = f"{primary.channel} {primary.audience}" matches = self._catalog.match_partners(tokens, focus, limit) if not matches: return [] avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.2 avg_ctr = mean([score.metrics.expected_ctr for score in campaigns]) if campaigns else 0.03 avg_confidence = ( mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.55 ) reach_multiplier = max(1.1, 1 + (avg_confidence - 0.5)) recs: List[InfluencerRecommendation] = [] for index, (match_score, partner) in enumerate(matches): collab_formats = partner.collab_formats or ["sponsored integration"] collab_format = collab_formats[index % len(collab_formats)] estimated_reach = int( round(partner.audience_size * partner.engagement_rate * reach_multiplier) ) estimated_reach = max(estimated_reach, int(partner.audience_size * 0.02)) lead_rate = max(0.015, avg_ctr * 1.15 + partner.engagement_rate / 3) projected_leads = max(12, int(round(estimated_reach * lead_rate))) investment = partner.avg_cpm * estimated_reach / 1000.0 revenue_per_lead = max(95.0, avg_roi * 120.0) projected_revenue = round(projected_leads * revenue_per_lead, 2) projected_roi = ( round(projected_revenue / investment, 2) if investment else round(avg_roi * 3, 2) ) keyword_matches = sorted( { token for token in tokens if token in {keyword.lower() for keyword in partner.keywords} } ) if keyword_matches: rationale = ( "Maps " + "/".join(keyword_matches[:3]) + f" intent into {partner.platform} {collab_format} with {partner.niche.lower()} authority." ) else: rationale = ( f"Pairs {partner.niche.lower()} reach with {partner.conversion_focus.lower()} focus " "to amplify campaign demand." ) if partner.case_study: rationale += f" Proof: {partner.case_study}" source = "influencer_api" if self.configured else "influencer_catalog" notes = None if not self.configured: notes = ( "Simulated from influencer partner corpus; set INFLUENCER_NETWORK_API_KEY to " "activate live network sourcing." ) recs.append( InfluencerRecommendation( name=partner.name, platform=partner.platform, niche=partner.niche, collab_format=collab_format, audience_size=partner.audience_size, engagement_rate=round(partner.engagement_rate, 4), estimated_reach=estimated_reach, projected_cpm=round(partner.avg_cpm, 2), projected_leads=projected_leads, projected_revenue=projected_revenue, projected_roi=projected_roi, rationale=rationale, source=source, notes=notes, ) ) return recs class AutomationSprintPlanner: """Builds a three-hour automation sprint tailored to the prompt context.""" def __init__( self, templates: Dict[str, Any] | None = None, defaults: Dict[str, Any] | None = None, ): data = templates or self._load_templates() self._templates = data.get("stages", {}) self._overlays = data.get("focus_overlays", {}) base_defaults = data.get("defaults", {}).copy() if defaults: base_defaults.update(defaults) self._defaults = base_defaults @staticmethod def _load_templates(path: str = AUTOMATION_SPRINTS_DATASET) -> Dict[str, Any]: with open(path, "r", encoding="utf-8") as handle: return json.load(handle) def _stage_templates(self, stage: str) -> List[Dict[str, Any]]: if stage in self._templates: return list(self._templates[stage]) if "conversion" in self._templates: return list(self._templates["conversion"]) for entries in self._templates.values(): if entries: return list(entries) return [] def _focus_overlays(self, focus: str) -> List[Dict[str, Any]]: lowered = focus.lower() for key, tasks in self._overlays.items(): if key in lowered: return list(tasks) return [] @staticmethod def _safe_format(template: str, context: Dict[str, str]) -> str: class _SafeDict(dict): def __missing__(self, key: str) -> str: # pragma: no cover - defensive fallback return "{" + key + "}" return template.format_map(_SafeDict(context)) @staticmethod def _campaign_context( campaigns: Sequence[CampaignScore], index: int ) -> Dict[str, str]: if campaigns: primary = campaigns[0].campaign reference = campaigns[index % len(campaigns)].campaign secondary = campaigns[1 % len(campaigns)].campaign if len(campaigns) > 1 else primary return { "channel": reference.channel, "offer": reference.offer, "audience": reference.audience, "objective": reference.objective, "primary_channel": primary.channel, "primary_offer": primary.offer, "primary_audience": primary.audience, "primary_objective": primary.objective, "support_channel": secondary.channel, "support_offer": secondary.offer, } return { "channel": "omnichannel automation", "offer": "flagship offer", "audience": "core buyers", "objective": "pipeline velocity", "primary_channel": "omnichannel", "primary_offer": "flagship offer", "primary_audience": "core buyers", "primary_objective": "pipeline velocity", "support_channel": "email", "support_offer": "activation play", } def plan( self, prompt: str, diagnostics: PromptDiagnostics, campaigns: Sequence[CampaignScore], limit: int, ) -> AutomationSprint: block = int(self._defaults.get("block_minutes", 30)) total_minutes = int(self._defaults.get("total_minutes", 180)) blocks = max(1, total_minutes // max(1, block)) templates = self._stage_templates(diagnostics.stage) overlays = self._focus_overlays(diagnostics.focus) if overlays: templates.extend(overlays) if not templates: templates = [ { "activity": "Audit acquisition funnel touchpoints", "outcome": "Document automation opportunities", "tooling": ["Notion"], "kpis": ["pipeline velocity"], "automation_level": "assist", } ] tasks: List[AutomationTask] = [] summary_drivers: List[str] = [] for idx in range(blocks): template = templates[idx % len(templates)] context = self._campaign_context(campaigns, idx) context["stage"] = diagnostics.stage context["focus"] = diagnostics.focus activity = self._safe_format(template.get("activity", "Execute automation sprint"), context) outcome = self._safe_format(template.get("outcome", "Drive measurable impact"), context) tooling = list(template.get("tooling", [])) kpis = [kpi.lower() for kpi in template.get("kpis", [])] automation_level = template.get("automation_level", "assist") start_minute = idx * block end_minute = start_minute + block source = self._defaults.get("source", "automation_playbooks") tasks.append( AutomationTask( start_minute=start_minute, end_minute=end_minute, activity=activity, outcome=outcome, tooling=tooling, kpis=kpis, automation_level=automation_level, source=source, ) ) channel = context.get("channel") if channel and channel not in summary_drivers: summary_drivers.append(channel) channels_summary = ", ".join(summary_drivers[:3]) or "omnichannel" summary = ( f"Three-hour automation sprint targeting {diagnostics.stage} intent for {diagnostics.focus} focus " f"anchored in {channels_summary} plays." ) notes = self._defaults.get("notes") return AutomationSprint( total_minutes=blocks * block, stage=diagnostics.stage, focus=diagnostics.focus, summary=summary, tasks=tasks, notes=notes, source=self._defaults.get("source", "automation_playbooks"), ) class MarketingIntelligenceEngine: """Ranks marketing campaigns against free-form prompts.""" def __init__( self, campaigns: Sequence[MarketingCampaign], benchmarks: MarketingBenchmarks | None = None, google_process: GoogleAdsProcess | None = None, cj_process: CJAffiliateProcess | None = None, crypto_process: CryptoMarketProcess | None = None, stock_process: StockMarketProcess | None = None, stripe_process: StripeProcess | None = None, paypal_process: PayPalProcess | None = None, payment_process: PaymentIntegrationProcess | None = None, investing_process: InvestingProcess | None = None, amazon_process: AmazonMarketplaceProcess | None = None, ebay_process: EbayMarketplaceProcess | None = None, aws_process: AWSRevenueProcess | None = None, budget_process: BudgetAllocatorProcess | None = None, automation_process: AutomationSprintPlanner | None = None, learning_process: LearningLoopProcess | None = None, meta_learning_process: MetaLearningProcess | None = None, competitive_process: CompetitiveIntelligenceProcess | None = None, influencer_process: InfluencerCollaborationProcess | None = None, scenario_process: "ScenarioPlanningProcess" | None = None, statistical_process: StatisticalReasoningProcess | None = None, cognitive_process: CognitiveArchitectureProcess | None = None, agentic_process: AgenticSystemsProcess | None = None, knowledge_process: KnowledgeWorkflowProcess | None = None, marketing_knowledge_process: MarketingKnowledgeProcess | None = None, code_process: CodeAutomationProcess | None = None, language_process: LanguageIntelligenceProcess | None = None, moe_process: MixtureOfExpertsProcess | None = None, agentic_loop_process: AgenticLoopProcess | None = None, quantum_process: QuantumSimulationProcess | None = None, physics_process: PhysicsEconomicsProcess | None = None, ): if not campaigns: raise ValueError("MarketingIntelligenceEngine requires at least one campaign") self._campaigns = list(campaigns) self._benchmarks = benchmarks or MarketingBenchmarks.from_file() self._google_process = google_process or GoogleAdsProcess() self._cj_process = cj_process or CJAffiliateProcess() self._crypto_process = crypto_process or CryptoMarketProcess() self._stock_process = stock_process or StockMarketProcess() self._stripe_process = stripe_process or StripeProcess() self._paypal_process = paypal_process or PayPalProcess() self._payment_process = payment_process or PaymentIntegrationProcess() self._investing_process = investing_process or InvestingProcess() self._amazon_process = amazon_process or AmazonMarketplaceProcess() self._ebay_process = ebay_process or EbayMarketplaceProcess() self._aws_process = aws_process or AWSRevenueProcess() self._budget_process = budget_process or BudgetAllocatorProcess() self._automation_process = automation_process or AutomationSprintPlanner() self._learning_process = learning_process or LearningLoopProcess() self._meta_learning_process = meta_learning_process or MetaLearningProcess() self._competitive_process = competitive_process or CompetitiveIntelligenceProcess() self._influencer_process = influencer_process or InfluencerCollaborationProcess() self._scenario_process = scenario_process or ScenarioPlanningProcess() self._statistical_process = statistical_process or StatisticalReasoningProcess() self._cognitive_process = cognitive_process or CognitiveArchitectureProcess() self._agentic_process = agentic_process or AgenticSystemsProcess() self._knowledge_process = knowledge_process or KnowledgeWorkflowProcess() self._marketing_knowledge_process = ( marketing_knowledge_process or MarketingKnowledgeProcess() ) self._code_process = code_process or CodeAutomationProcess() self._language_process = language_process or LanguageIntelligenceProcess() self._moe_process = moe_process or MixtureOfExpertsProcess() self._agentic_loop_process = agentic_loop_process or AgenticLoopProcess() self._quantum_process = quantum_process or QuantumSimulationProcess() self._physics_process = physics_process or PhysicsEconomicsProcess() @classmethod def from_file( cls, path: str = MARKETING_ENUMERATION, benchmark_path: str = MARKETING_BENCHMARKS, ) -> "MarketingIntelligenceEngine": lines = _load_lines(path) campaigns = [MarketingCampaign.parse(line) for line in lines] benchmarks = MarketingBenchmarks.from_file(benchmark_path) return cls(campaigns, benchmarks) @staticmethod def _keyword_set(prompt: str) -> List[str]: tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower()) filtered = [token for token in tokens if len(token) > 2] return filtered or ["campaign"] def recommend(self, prompt: str, limit: int = 3) -> List[MarketingCampaign]: tokens = self._keyword_set(prompt) def score(campaign: MarketingCampaign) -> tuple[int, int, str]: haystacks = { "channel": campaign.channel.lower(), "audience": campaign.audience.lower(), "offer": campaign.offer.lower(), "objective": campaign.objective.lower(), } weighted = 0 matched_features: set[str] = set() for token in tokens: for feature, text in haystacks.items(): if token in text: matched_features.add(feature) weight = 4 if feature == "audience" else 3 if feature == "offer" else 2 weighted += weight # fallback to slight preference for broader coverage even if nothing matched coverage_bonus = len(matched_features) return (-(weighted + coverage_bonus), len(campaign.raw), campaign.raw) ranked = sorted(self._campaigns, key=score) return ranked[: max(1, limit)] def recommend_statistical(self, prompt: str, limit: int = 3) -> List[CampaignScore]: tokens = self._keyword_set(prompt) scored: List[CampaignScore] = [] for campaign in self._campaigns: metrics = self._benchmarks.metrics_for(campaign) keyword_hits: Dict[str, int] = {"channel": 0, "audience": 0, "offer": 0, "objective": 0} haystacks = { "channel": campaign.channel.lower(), "audience": campaign.audience.lower(), "offer": campaign.offer.lower(), "objective": campaign.objective.lower(), } match_bonus = 0.0 for token in tokens: for feature, text in haystacks.items(): if token in text: keyword_hits[feature] += 1 match_bonus += 0.35 roi_component = metrics.expected_roi * 0.6 ctr_component = metrics.expected_ctr * 80.0 * 0.25 confidence_component = metrics.confidence * 10.0 * 0.1 keyword_component = match_bonus composite = roi_component + ctr_component + confidence_component + keyword_component scored.append( CampaignScore( campaign=campaign, score=composite, metrics=metrics, keyword_coverage=keyword_hits, ) ) ranked = sorted(scored, key=lambda entry: entry.score, reverse=True) return ranked[: max(1, limit)] def summarize(self, prompt: str, campaigns: Sequence[MarketingCampaign]) -> str: tokens = self._keyword_set(prompt) highlights = ", ".join(sorted(set(tokens))) count = len(campaigns) return f"Surface {count} campaign plays tuned for {highlights}." @staticmethod def _infer_stage(tokens: Sequence[str]) -> Tuple[str, List[str]]: mapping = { "awareness": {"awareness", "reach", "impressions", "brand", "launch", "attention"}, "consideration": {"consideration", "education", "webinar", "nurture", "pipeline", "engagement"}, "conversion": { "conversion", "roi", "ctr", "leads", "demos", "trial", "customers", "acquisition", }, "retention": {"retention", "renewal", "expansion", "upsell", "loyalty", "churn"}, } hits: Dict[str, List[str]] = {key: [] for key in mapping} for token in tokens: for stage, keywords in mapping.items(): if token in keywords: hits[stage].append(token) stage = max(mapping, key=lambda key: len(hits[key])) if all(len(values) == 0 for values in hits.values()): stage = "conversion" return stage, hits[stage] @staticmethod def _infer_urgency(prompt: str) -> str: lowered = prompt.lower() if any(keyword in lowered for keyword in ["urgent", "asap", "immediately", "right away", "now"]): return "urgent" if any(keyword in lowered for keyword in ["this quarter", "q4", "deadline", "accelerate"]): return "time-sensitive" return "standard" @staticmethod def _infer_focus(tokens: Sequence[str]) -> Tuple[str, List[str]]: focus_buckets = { "enterprise b2b": {"enterprise", "cxo", "c-suite", "fortune", "b2b", "saas"}, "startup founders": {"founder", "startup", "scaleup", "seed", "series"}, "consumer": {"consumer", "b2c", "retail", "shopper", "app"}, "developer": {"developer", "devops", "engineer", "technical", "api"}, } hits: Dict[str, List[str]] = {bucket: [] for bucket in focus_buckets} for token in tokens: for focus, keywords in focus_buckets.items(): if token in keywords: hits[focus].append(token) focus = max(focus_buckets, key=lambda key: len(hits[key])) if len(hits[focus]) == 0: return "broad b2b", [] return focus, hits[focus] @staticmethod def _recommend_kpis(stage: str) -> List[str]: mapping = { "awareness": ["impressions", "share of voice", "reach growth"], "consideration": ["content engagement", "marketing qualified leads", "pipeline velocity"], "conversion": ["sales qualified leads", "demo-to-close rate", "customer acquisition cost"], "retention": ["net revenue retention", "renewal rate", "expansion ARR"], } return mapping.get(stage, ["pipeline velocity", "conversion rate"]) @staticmethod def _recommend_opportunities(stage: str, focus: str) -> List[str]: opportunities = { "awareness": [ "Anchor thought-leadership content to prime the market", "Deploy programmatic amplification across hero stories", ], "consideration": [ "Introduce nurture sequences aligned to buyer pains", "Layer intent scoring to prioritize engaged accounts", ], "conversion": [ "Bundle conversion offers with social proof assets", "Tighten sales enablement loops with marketing triggers", ], "retention": [ "Launch customer marketing to reinforce product value", "Pair success cadences with expansion playbooks", ], } base = opportunities.get(stage, []) if focus.startswith("enterprise"): base = base + ["Align executive-level narratives with ROI benchmarks"] elif "startup" in focus: base = base + ["Stand up scrappy experiments to validate fast wins"] elif "consumer" in focus: base = base + ["Personalize lifecycle messaging using behavioral cohorts"] return base @staticmethod def _detect_risks(prompt: str) -> List[str]: lowered = prompt.lower() risks: List[str] = [] if "flat" in lowered or "stalled" in lowered or "plateau" in lowered: risks.append("Performance plateau signaled") if "limited budget" in lowered or "tight budget" in lowered: risks.append("Budget constraints highlighted") if "new market" in lowered or "unknown" in lowered: risks.append("Market unfamiliarity risk") if "churn" in lowered or "retention" in lowered: risks.append("Retention fragility noted") return risks def diagnose(self, prompt: str) -> PromptDiagnostics: tokens = self._keyword_set(prompt) stage, stage_hits = self._infer_stage(tokens) urgency = self._infer_urgency(prompt) focus, focus_hits = self._infer_focus(tokens) kpis = self._recommend_kpis(stage) opportunities = self._recommend_opportunities(stage, focus) risks = self._detect_risks(prompt) keyword_summary = { "intent": stage_hits, "focus": focus_hits, "tokens": tokens, } confidence_seed = len(stage_hits) + len(focus_hits) confidence = max(0.35, min(0.95, 0.35 + 0.1 * confidence_seed)) return PromptDiagnostics( stage=stage, urgency=urgency, focus=focus, recommended_kpis=kpis, opportunities=opportunities, risk_signals=risks, keyword_summary=keyword_summary, confidence=round(confidence, 2), ) def portfolio(self, prompt: str, campaigns: Sequence[CampaignScore]) -> CampaignMetrics: if not campaigns: raise ValueError("portfolio requires at least one campaign score") roi = sum(entry.metrics.expected_roi for entry in campaigns) / len(campaigns) ctr = sum(entry.metrics.expected_ctr for entry in campaigns) / len(campaigns) confidence = sum(entry.metrics.confidence for entry in campaigns) / len(campaigns) drivers: Dict[str, float] = {} for entry in campaigns: for feature, value in entry.metrics.drivers.items(): drivers[feature] = max(drivers.get(feature, 0.0), value) return CampaignMetrics( expected_roi=roi, expected_ctr=ctr, confidence=confidence, drivers=drivers, ) def keywords(self, prompt: str) -> List[str]: return self._keyword_set(prompt) def statistical_reasoning_summary( self, prompt: str, campaigns: Sequence[CampaignScore], *, budget: BudgetPlan | None, stock_markets: Sequence[StockMarketForecast] | None, crypto_markets: Sequence[CryptoMarketSnapshot] | None, limit: int, ) -> StatisticalReasoningSummary: return self._statistical_process.analyze( prompt, campaigns, budget=budget, stock_markets=stock_markets, crypto_markets=crypto_markets, limit=limit, ) def expansion_grid(self, prompt: str, campaigns: Sequence[CampaignScore]) -> List[GrowthExperiment]: """Design a set of Experiment Growth Grid (EGG) plays from campaign rankings.""" if not campaigns: raise ValueError("expansion_grid requires at least one campaign score") diagnostics = self.diagnose(prompt) stage_recipes: Dict[str, List[Dict[str, str]]] = { "awareness": [ { "name": "Narrative Amplification Sprint", "hypothesis": ( "If we spotlight {offer} through {channel}, " "we will warm {audience} toward {objective}." ), "play": "Launch hero content and amplify across paid/owned reach streams.", "metric": "impressions", }, { "name": "Thought Leadership Cascade", "hypothesis": ( "Positioning {audience} challenges alongside {offer} " "via {channel} will expand top-of-funnel attention." ), "play": "Sequence authoritative narratives across social and partner syndication.", "metric": "share of voice", }, ], "consideration": [ { "name": "Intent Magnet Program", "hypothesis": ( "If we orchestrate {offer} journeys through {channel}, " "we can mature {audience} into evaluators for {objective}." ), "play": "Build nurture modules triggered by behavioral intent signals.", "metric": "marketing qualified leads", }, { "name": "Education Sprint", "hypothesis": ( "Providing {audience} with {offer} across {channel} " "will compress evaluation cycles." ), "play": "Layer webinars, playbooks, and retargeting to reinforce learnings.", "metric": "content engagement", }, ], "conversion": [ { "name": "Pipeline Acceleration Lab", "hypothesis": ( "If {audience} experience {offer} via {channel}, " "we can unlock {objective} faster." ), "play": "Run multi-touch conversion blitz with calibrated social proof.", "metric": "sales qualified leads", }, { "name": "Revenue Experiment Track", "hypothesis": ( "Blending {offer} within {channel} for {audience} will uplift deal velocity." ), "play": "Pair high-intent triggers with offer experimentation and CRO support.", "metric": "conversion rate", }, ], "retention": [ { "name": "Value Expansion Studio", "hypothesis": ( "If {audience} engage with {offer} programs via {channel}, " "we will fortify {objective}." ), "play": "Launch customer storytelling and advocacy loops to reinforce value.", "metric": "renewal rate", }, { "name": "Lifecycle Momentum Cell", "hypothesis": ( "Activating {offer} through {channel} for {audience} " "will catalyze expansion outcomes." ), "play": "Coordinate success-led campaigns with product usage nudges.", "metric": "net revenue retention", }, ], } focus_overlays: Dict[str, Dict[str, str]] = { "enterprise b2b": { "play": "Map executive-specific proof points and ABM orchestration layers.", "metric": "executive meeting volume", }, "startup founders": { "play": "Prototype scrappy experiments with rapid iteration cadences.", "metric": "activation-to-demo ratio", }, "consumer": { "play": "Employ personalization matrices and lifecycle messaging bursts.", "metric": "repeat purchase rate", }, "developer": { "play": "Enable technical champions with sandboxes, tutorials, and community proof.", "metric": "trial-to-adoption rate", }, "default": { "play": "Instrument experiment tracking with weekly conversion retros.", "metric": "conversion rate", }, } base_recipes = stage_recipes.get(diagnostics.stage, stage_recipes["conversion"]) overlay = focus_overlays.get(diagnostics.focus, focus_overlays["default"]) recommended = diagnostics.recommended_kpis or ["conversion rate"] experiments: List[GrowthExperiment] = [] for index, scored in enumerate(campaigns[: max(1, len(campaigns))]): recipe = base_recipes[index % len(base_recipes)] name = f"{recipe['name']} via {scored.campaign.channel.title()}" hypothesis = recipe["hypothesis"].format( offer=scored.campaign.offer, channel=scored.campaign.channel, audience=scored.campaign.audience, objective=scored.campaign.objective, ) playbook = [ f"Activate {scored.campaign.channel} targeting {scored.campaign.audience}.", recipe["play"], overlay["play"], f"Align offer '{scored.campaign.offer}' to accelerate {scored.campaign.objective} outcomes.", ] primary_metric = recipe.get("metric", recommended[0]) secondary_candidates = set(recommended[1:]) secondary_candidates.add(overlay["metric"]) secondary_metrics = sorted(secondary_candidates) confidence = (diagnostics.confidence + scored.metrics.confidence) / 2 experiments.append( GrowthExperiment( name=name, hypothesis=hypothesis, primary_metric=primary_metric, secondary_metrics=secondary_metrics, playbook=playbook, confidence=round(max(0.2, min(0.99, confidence)), 2), ) ) return experiments def google_ads_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[GoogleCampaignPerformance]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._google_process.performance(prompt, scored, limit) def cj_affiliate_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[CJAffiliateRecommendation]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._cj_process.offers(prompt, scored, limit) def crypto_market_enrichment( self, prompt: str, limit: int = 3, ) -> List[CryptoMarketSnapshot]: return self._crypto_process.market_data(prompt, limit) def stock_market_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[StockMarketForecast]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._stock_process.forecasts(prompt, scored, limit) def stripe_checkout_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[StripeMonetizationPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._stripe_process.plans(prompt, scored, limit) def paypal_checkout_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[PayPalMonetizationPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._paypal_process.plans(prompt, scored, limit) def payment_integration_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 4, ) -> List[PaymentIntegrationPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._payment_process.plans(prompt, scored, limit) def investing_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[InvestingPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._investing_process.plans(prompt, scored, limit) def physics_economics_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[PhysicsEconomicsPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._physics_process.plans(prompt, scored, limit) def influencer_collaboration_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[InfluencerRecommendation]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._influencer_process.recommendations(prompt, scored, limit) def amazon_marketplace_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[AmazonProductRecommendation]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._amazon_process.listings(prompt, scored, limit) def ebay_marketplace_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[EbayRevenuePlay]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._ebay_process.listings(prompt, scored, limit) def aws_cloud_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[AWSRevenueRecommendation]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._aws_process.plays(prompt, scored, limit) def budget_optimizer( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> BudgetPlan: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return BudgetPlan( total_budget=0.0, allocations=[], expected_revenue=0.0, expected_roi=0.0, payback_period_weeks=0.0, summary="No campaigns available for budget allocation.", ) return self._budget_process.plan(prompt, scored, limit) def automation_sprint( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> AutomationSprint: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) diagnostics = self.diagnose(prompt) return self._automation_process.plan(prompt, diagnostics, scored, limit) def competitive_landscape( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[CompetitiveSignal]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) diagnostics = self.diagnose(prompt) return self._competitive_process.signals(prompt, diagnostics, scored, limit) def learning_loop( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[LearningLoopRecommendation]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) diagnostics = self.diagnose(prompt) return self._learning_process.recommend(prompt, diagnostics, scored, limit) def cognitive_architecture_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[CognitiveCapabilityPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._cognitive_process.plans(prompt, scored, limit) def agentic_system_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[AgenticSystemPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._agentic_process.plans(prompt, scored, limit) def knowledge_workflow_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[KnowledgeWorkflowPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._knowledge_process.plans(prompt, scored, limit) def marketing_knowledge_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[MarketingKnowledgePlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._marketing_knowledge_process.plans(prompt, scored, limit) def language_capability_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[LanguageCapabilityPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._language_process.plans(prompt, scored, limit) def code_automation_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[CodeAutomationPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._code_process.plans(prompt, scored, limit) def meta_learning_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[MetaLearningPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._meta_learning_process.plans(prompt, scored, limit) def rag_pipeline_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[KnowledgeWorkflowPlan]: plans = self.knowledge_workflow_enrichment(prompt, campaigns, limit) return [plan for plan in plans if plan.category == "rag"] def knowledge_graph_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[KnowledgeWorkflowPlan]: plans = self.knowledge_workflow_enrichment(prompt, campaigns, limit) return [plan for plan in plans if plan.category == "kag"] def mixture_of_experts_activation( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[MixtureOfExpertsPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._moe_process.activate(prompt, scored, limit) def agentic_loop_blueprints( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 1, ) -> List[AgenticLoopPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, max(1, limit)) if not scored: return [] return self._agentic_loop_process.plans(prompt, scored, limit) def quantum_simulation_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 1, ) -> List[QuantumStressTestPlan]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, max(1, limit)) if not scored: return [] return self._quantum_process.plans(prompt, scored, limit) def scenario_planning( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, *, limit: int = 3, budget: BudgetPlan | None = None, google_ads: Sequence[GoogleCampaignPerformance] | None = None, cj_affiliate: Sequence[CJAffiliateRecommendation] | None = None, amazon: Sequence[AmazonProductRecommendation] | None = None, ebay: Sequence[EbayRevenuePlay] | None = None, aws: Sequence[AWSRevenueRecommendation] | None = None, stripe: Sequence[StripeMonetizationPlan] | None = None, paypal: Sequence[PayPalMonetizationPlan] | None = None, payments: Sequence[PaymentIntegrationPlan] | None = None, investing: Sequence[InvestingPlan] | None = None, crypto: Sequence[CryptoMarketSnapshot] | None = None, influencers: Sequence[InfluencerRecommendation] | None = None, stocks: Sequence[StockMarketForecast] | None = None, statistics: StatisticalReasoningSummary | None = None, cognitive: Sequence[CognitiveCapabilityPlan] | None = None, agentic: Sequence[AgenticSystemPlan] | None = None, knowledge: Sequence[KnowledgeWorkflowPlan] | None = None, marketing_knowledge: Sequence[MarketingKnowledgePlan] | None = None, code_automation: Sequence[CodeAutomationPlan] | None = None, language: Sequence[LanguageCapabilityPlan] | None = None, meta_learning: Sequence[MetaLearningPlan] | None = None, moe: Sequence[MixtureOfExpertsPlan] | None = None, agentic_loops: Sequence[AgenticLoopPlan] | None = None, quantum: Sequence[QuantumStressTestPlan] | None = None, physics_economics: Sequence[PhysicsEconomicsPlan] | None = None, ) -> ScenarioAnalysis: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) diagnostics = self.diagnose(prompt) return self._scenario_process.analyze( prompt, diagnostics, scored, budget=budget, google_ads=google_ads or [], cj_affiliate=cj_affiliate or [], amazon=amazon or [], ebay=ebay or [], aws=aws or [], stripe=stripe or [], paypal=paypal or [], payments=payments or [], investing=investing or [], crypto=crypto or [], influencers=influencers or [], stocks=stocks or [], statistics=statistics, cognitive=cognitive or [], agentic=agentic or [], knowledge=knowledge or [], marketing_knowledge=marketing_knowledge or [], code_automation=code_automation or [], language=language or [], meta_learning=meta_learning or [], moe=moe or [], agentic_loops=agentic_loops or [], quantum=quantum or [], physics_economics=physics_economics or [], limit=limit, ) def excellence_summary( self, prompt: str, campaigns: Sequence[CampaignScore], *, limit: int, rag: Sequence[MarketingCampaign], egg: Sequence[GrowthExperiment], google_ads: Sequence[GoogleCampaignPerformance], cj_affiliate: Sequence[CJAffiliateRecommendation], crypto_markets: Sequence[CryptoMarketSnapshot], stripe: Sequence[StripeMonetizationPlan], paypal: Sequence[PayPalMonetizationPlan], payments: Sequence[PaymentIntegrationPlan], investing: Sequence[InvestingPlan], amazon: Sequence[AmazonProductRecommendation], ebay: Sequence[EbayRevenuePlay], aws: Sequence[AWSRevenueRecommendation], budget: BudgetPlan | None, automation: AutomationSprint | None, learning_loop: Sequence[LearningLoopRecommendation], competitive: Sequence[CompetitiveSignal], influencers: Sequence[InfluencerRecommendation], stocks: Sequence[StockMarketForecast], statistics: StatisticalReasoningSummary | None = None, cognitive: Sequence[CognitiveCapabilityPlan] | None = None, agentic: Sequence[AgenticSystemPlan] | None = None, knowledge: Sequence[KnowledgeWorkflowPlan] | None = None, marketing_knowledge: Sequence[MarketingKnowledgePlan] | None = None, code_automation: Sequence[CodeAutomationPlan] | None = None, language: Sequence[LanguageCapabilityPlan] | None = None, meta_learning: Sequence[MetaLearningPlan] | None = None, moe: Sequence[MixtureOfExpertsPlan] | None = None, agentic_loops: Sequence[AgenticLoopPlan] | None = None, quantum: Sequence[QuantumStressTestPlan] | None = None, physics_economics: Sequence[PhysicsEconomicsPlan] | None = None, ) -> ExcellenceSummary: limit = max(1, limit) roi_avg = mean(entry.metrics.expected_roi for entry in campaigns) if campaigns else 0.0 ctr_avg = mean(entry.metrics.expected_ctr for entry in campaigns) if campaigns else 0.0 confidence_avg = ( mean(entry.metrics.confidence for entry in campaigns) if campaigns else 0.0 ) roi_norm = min(1.0, max(0.0, roi_avg / 5.0)) ctr_norm = min(1.0, max(0.0, ctr_avg / 0.05)) confidence_norm = min(1.0, max(0.0, confidence_avg)) coverage_map: Dict[str, float] = { "rag": min(1.0, len(rag) / limit), "egg": min(1.0, len(egg) / limit), "google_ads": min(1.0, len(google_ads) / limit), "cj_affiliate": min(1.0, len(cj_affiliate) / limit), "crypto_markets": min(1.0, len(crypto_markets) / limit), "stock_markets": min(1.0, len(stocks) / limit), "stripe": min(1.0, len(stripe) / limit), "paypal": min(1.0, len(paypal) / limit), "payment_integrations": min(1.0, len(payments) / limit), "investing": min(1.0, len(investing) / limit), "influencer_collaborations": min(1.0, len(influencers) / limit), "amazon_marketplace": min(1.0, len(amazon) / limit), "ebay_marketplace": min(1.0, len(ebay) / limit), "aws_cloud": min(1.0, len(aws) / limit), "learning_loop": min(1.0, len(learning_loop) / limit), "competitive_landscape": min(1.0, len(competitive) / limit), } coverage_map["budget_optimizer"] = 1.0 if budget and budget.allocations else 0.0 coverage_map["automation"] = 1.0 if automation and automation.tasks else 0.0 coverage_map["statistical_reasoning"] = 1.0 if statistics else 0.0 coverage_map["cognitive_architecture"] = ( min(1.0, len(cognitive or []) / limit) if cognitive is not None else 0.0 ) coverage_map["agentic_systems"] = ( min(1.0, len(agentic or []) / limit) if agentic is not None else 0.0 ) coverage_map["code_automation"] = ( min(1.0, len(code_automation or []) / limit) if code_automation is not None else 0.0 ) coverage_map["marketing_knowledge"] = ( min(1.0, len(marketing_knowledge or []) / limit) if marketing_knowledge is not None else 0.0 ) coverage_map["meta_learning"] = ( min(1.0, len(meta_learning or []) / limit) if meta_learning is not None else 0.0 ) rag_workflows = [plan for plan in (knowledge or []) if plan.category == "rag"] kag_workflows = [plan for plan in (knowledge or []) if plan.category == "kag"] coverage_map["rag_workflows"] = min(1.0, len(rag_workflows) / limit) if knowledge is not None else 0.0 coverage_map["knowledge_graphs"] = min(1.0, len(kag_workflows) / limit) if knowledge is not None else 0.0 coverage_map["quantum_simulation"] = ( min(1.0, len(quantum or []) / limit) if quantum is not None else 0.0 ) coverage_map["language_intelligence"] = ( min(1.0, len(language or []) / limit) if language is not None else 0.0 ) coverage_map["mixture_of_experts"] = 1.0 if moe and any(plan.activations for plan in moe) else 0.0 coverage_map["agentic_loops"] = 1.0 if agentic_loops else 0.0 coverage_map["physics_economics"] = ( min(1.0, len(physics_economics or []) / limit) if physics_economics is not None else 0.0 ) coverage_norm = mean(coverage_map.values()) if coverage_map else 0.0 excellence_index = round( (roi_norm * 0.35 + ctr_norm * 0.2 + confidence_norm * 0.25 + coverage_norm * 0.2) * 100, 2, ) if excellence_index >= 85: maturity = "world-class" elif excellence_index >= 70: maturity = "elite" elif excellence_index >= 55: maturity = "scaling" else: maturity = "emerging" keywords = self.keywords(prompt)[:3] if keywords: headline_focus = ", ".join(keywords) else: headline_focus = "core growth themes" headline = ( f"{maturity.title()} marketing ANI readiness anchored in {headline_focus}." ) strengths: List[str] = [] if roi_norm >= 0.75: strengths.append("Benchmark ROI trajectory rivals top performers.") elif roi_norm >= 0.5: strengths.append("ROI projections support confident scaling.") if coverage_norm >= 0.7: strengths.append("Process coverage spans key enrichment workflows.") if len(competitive) >= 1: strengths.append("Competitive counter-moves actively inform campaign plans.") if payments: payment_plan = payments[0] strengths.append( f"{payment_plan.provider} integration operating at {payment_plan.maturity} maturity accelerates revenue capture." ) if investing: investing_plan = investing[0] strengths.append( f"Investing platform {investing_plan.provider} ({investing_plan.maturity}) opens financial product monetization." ) if len(influencers) >= 1: strengths.append("Influencer collaborations unlock rapid audience amplification.") if stocks: top_stock = stocks[0] if top_stock.forecast_return_90d > 0.1: strengths.append( f"{top_stock.name} outlook adds capital market tailwind to marketing scale." ) if statistics and statistics.success_probability >= 0.75: strengths.append( f"Statistical reasoning signals {statistics.success_probability:.0%} success probability." ) if cognitive: top_capability = cognitive[0] if top_capability.maturity in {"advanced", "autonomous"}: strengths.append( f"Cognitive architecture running {top_capability.maturity} on {top_capability.capability}." ) if language: top_language = language[0] if top_language.maturity in {"advanced", "autonomous"}: strengths.append( f"Language intelligence {top_language.capability} {top_language.maturity} for conversational monetization." ) if code_automation: code_plan = code_automation[0] if code_plan.maturity in {"advanced", "autonomous"}: strengths.append( f"Code automation {code_plan.capability} operating at {code_plan.maturity} maturity stabilizes releases." ) if meta_learning: meta_plan = meta_learning[0] if meta_plan.maturity in {"advanced", "autonomous", "elite"}: strengths.append( f"Meta-learning {meta_plan.capability} ({meta_plan.maturity}) keeps adaptive improvements compounding." ) if agentic: system = agentic[0] strengths.append( f"Agentic system {system.system_name} orchestrates {system.maturity} tool calling stack." ) if knowledge: strengths.append("Knowledge workflows keep retrieval and reasoning grounded for every prompt.") if moe and moe[0].activations: strengths.append("Mixture-of-experts router balances revenue experts for routing confidence.") if agentic_loops: strengths.append("Agentic loop blueprint sustains continuous perception→action execution.") if quantum: top_plan = quantum[0] strengths.append( f"Quantum simulator {top_plan.simulator} ({top_plan.maturity}) validates cross-provider stress scenarios." ) if physics_economics: physics_plan = physics_economics[0] strengths.append( f"Physics/economics fusion {physics_plan.capability} ({physics_plan.maturity}) fortifies complex revenue models." ) watchouts: List[str] = [] if ctr_norm < 0.4: watchouts.append( "Click-through benchmarks trail aggressive targets; iterate creative experiments." ) if confidence_norm < 0.6: watchouts.append("Benchmark confidence is moderate; collect more lift validation.") if coverage_norm < 0.5: watchouts.append( "Several enrichment processes returned limited data; review integration readiness." ) if len(influencers) == 0: watchouts.append("Source influencer partners to extend reach beyond owned channels.") if stocks and mean(forecast.volatility for forecast in stocks) > 0.25: watchouts.append("Equity volatility elevated; stage reinvestment with safeguards.") if statistics and statistics.success_probability < 0.6: watchouts.append( "Statistical reasoning indicates moderate success probability; reinforce testing cadence." ) if not cognitive: watchouts.append( "Activate cognitive architecture plans so reasoning and learning stay orchestrated." ) if not language: watchouts.append( "Stand up language intelligence workflows so prompts, chats, and copy stay revenue-ready." ) if not code_automation: watchouts.append( "Instrument AI code fixing and intelligent automation so shipping velocity stays revenue-safe." ) if not meta_learning: watchouts.append( "Stand up meta-learning loops so experiments keep getting smarter between cycles." ) if not payments: watchouts.append("Stand up payment provider runbooks so monetization launches stay unblockable.") if not investing: watchouts.append("Add investing partners to unlock financial services revenue streams.") if not agentic: watchouts.append("Wire agentic systems to automate tool calling and database access.") if not knowledge: watchouts.append("Stand up RAG/KAG pipelines to ground responses in curated knowledge.") if not moe: watchouts.append("Calibrate mixture-of-experts routing to blend marketplace monetization signals.") if not agentic_loops: watchouts.append("Document an agentic loop so automation can self-correct between cycles.") if not quantum: watchouts.append("Activate quantum stress testing to pressure test markets and payments in tandem.") if not physics_economics: watchouts.append("Stand up physics/economics orchestration so complex market mechanics stay resilient.") if not strengths: strengths.append("Foundational datasets available for orchestration.") if not watchouts: watchouts.append("Continue monitoring live integrations to maintain excellence.") notes = ( "Excellence index synthesizes ROI, CTR, confidence, and process coverage heuristics." ) if stocks: top_stock = stocks[0] notes += ( f" | Stock outlook {top_stock.symbol} {top_stock.trend} with " f"{top_stock.forecast_return_30d*100:.1f}% 30d move." ) if statistics: notes += f" | {statistics.narrative}" if cognitive: notes += ( f" | Cognitive architecture anchored by {cognitive[0].capability} ({cognitive[0].maturity})." ) if agentic: notes += ( f" | Agentic system {agentic[0].system_name} ({agentic[0].maturity}) governing tool orchestration." ) if knowledge: notes += ( f" | Knowledge workflows cover {len(rag_workflows)} RAG and {len(kag_workflows)} KAG pipelines." ) if marketing_knowledge: notes += ( f" | Marketing knowledge sync {marketing_knowledge[0].name} ({marketing_knowledge[0].maturity})." ) if language: notes += ( f" | Language workflows anchor {language[0].capability} ({language[0].maturity})." ) if meta_learning: notes += ( f" | Meta-learning loops led by {meta_learning[0].capability} ({meta_learning[0].maturity})." ) if code_automation: notes += ( f" | Code automation {code_automation[0].capability} ({code_automation[0].maturity}) protects deployments." ) if payments: notes += ( f" | Payment integrations led by {payments[0].provider} ({payments[0].maturity})." ) if investing: notes += ( f" | Investing runway anchored by {investing[0].provider} ({investing[0].maturity})." ) if moe: activation_count = len(moe[0].activations) if moe and moe[0].activations else 0 notes += f" | MoE router activated {activation_count} revenue experts." if agentic_loops: notes += f" | Agentic loop {agentic_loops[0].name} maintains continuous execution." if quantum: notes += ( f" | Quantum simulator {quantum[0].simulator} covers {len(quantum[0].stress_tests)} stress scenarios." ) if physics_economics: notes += ( f" | Physics/economics plan {physics_economics[0].capability} ({physics_economics[0].maturity}) anchors complex mechanics." ) return ExcellenceSummary( excellence_index=excellence_index, maturity_level=maturity, headline=headline, strengths=strengths, watchouts=watchouts, process_coverage=coverage_map, notes=notes, ) def executive_playbook( self, prompt: str, campaigns: Sequence[CampaignScore], *, limit: int, rag: Sequence[MarketingCampaign], egg: Sequence[GrowthExperiment], google_ads: Sequence[GoogleCampaignPerformance], cj_affiliate: Sequence[CJAffiliateRecommendation], amazon: Sequence[AmazonProductRecommendation], ebay: Sequence[EbayRevenuePlay], paypal: Sequence[PayPalMonetizationPlan], payments: Sequence[PaymentIntegrationPlan] | None = None, investing: Sequence[InvestingPlan] | None = None, budget: BudgetPlan | None, automation: AutomationSprint | None, learning_loop: Sequence[LearningLoopRecommendation], competitive: Sequence[CompetitiveSignal], influencers: Sequence[InfluencerRecommendation], stocks: Sequence[StockMarketForecast], excellence: ExcellenceSummary | None = None, statistics: StatisticalReasoningSummary | None = None, cognitive: Sequence[CognitiveCapabilityPlan] | None = None, agentic: Sequence[AgenticSystemPlan] | None = None, knowledge: Sequence[KnowledgeWorkflowPlan] | None = None, marketing_knowledge: Sequence[MarketingKnowledgePlan] | None = None, code_automation: Sequence[CodeAutomationPlan] | None = None, language: Sequence[LanguageCapabilityPlan] | None = None, meta_learning: Sequence[MetaLearningPlan] | None = None, moe: Sequence[MixtureOfExpertsPlan] | None = None, agentic_loops: Sequence[AgenticLoopPlan] | None = None, quantum: Sequence[QuantumStressTestPlan] | None = None, physics_economics: Sequence[PhysicsEconomicsPlan] | None = None, ) -> ExecutivePlaybook: diagnostics = self.diagnose(prompt) def _channel_key(text: str) -> str: return re.sub(r"\s+", " ", text.lower().strip()) limit = max(1, limit) top_scores = list(campaigns[: limit if campaigns else 1]) if not top_scores and campaigns: top_scores = list(campaigns) experiment_map: Dict[str, GrowthExperiment] = {} for experiment in egg: name_lower = experiment.name.lower() channel_fragment: str | None = None if " via " in name_lower: channel_fragment = name_lower.split(" via ", 1)[1].strip() if (not channel_fragment or not channel_fragment.strip()) and experiment.playbook: first_step = experiment.playbook[0].lower() match = re.search(r"activate (.+?) targeting", first_step) if match: channel_fragment = match.group(1).strip() if channel_fragment: experiment_map.setdefault(_channel_key(channel_fragment), experiment) budget_map: Dict[str, BudgetAllocation] = {} if budget: for allocation in budget.allocations: budget_map[_channel_key(allocation.channel)] = allocation learning_map: Dict[str, List[LearningLoopRecommendation]] = {} for recommendation in learning_loop: channel_hint = recommendation.channel or recommendation.stage or recommendation.play_name key = _channel_key(channel_hint) learning_map.setdefault(key, []).append(recommendation) competitive_map: Dict[str, List[CompetitiveSignal]] = {} for signal in competitive: competitive_map.setdefault(_channel_key(signal.channel), []).append(signal) rag_channels = {idx: rag_campaign.channel for idx, rag_campaign in enumerate(rag)} influencer_pool = list(influencers) investing = list(investing or []) meta_learning = list(meta_learning or []) quantum = list(quantum or []) physics_economics = list(physics_economics or []) marketing_knowledge = list(marketing_knowledge or []) urgency_to_horizon = {"urgent": 72, "time-sensitive": 120, "standard": 168} base_horizon = urgency_to_horizon.get(diagnostics.urgency, 168) campaign_confidence = ( mean(score.metrics.confidence for score in top_scores) if top_scores else 0.5 ) roi_confidence = ( mean(score.metrics.expected_roi for score in top_scores) if top_scores else 0.5 ) budget_confidence = ( min(0.95, max(0.35, (budget.expected_roi / 5.0))) if budget else 0.55 ) learning_confidence = ( mean(rec.confidence for rec in learning_loop) if learning_loop else 0.55 ) automation_confidence = 0.75 if automation and automation.tasks else 0.5 statistics_confidence = statistics.confidence if statistics else 0.6 agentic_confidence = 0.78 if agentic else 0.5 knowledge_confidence = 0.76 if knowledge else 0.5 code_confidence = 0.76 if code_automation else 0.5 language_confidence = 0.76 if language else 0.5 meta_learning_confidence = 0.78 if meta_learning else 0.5 moe_confidence = 0.78 if moe and moe[0].activations else 0.5 loop_confidence = 0.8 if agentic_loops else 0.55 investing_confidence = 0.78 if investing else 0.5 payment_confidence = 0.78 if payments else 0.5 quantum_confidence = 0.78 if quantum else 0.5 physics_confidence = 0.78 if physics_economics else 0.5 overall_confidence = round( min( 0.99, max( 0.35, mean( [ diagnostics.confidence, campaign_confidence, min(0.95, max(0.3, roi_confidence / 5.0)), budget_confidence, learning_confidence, automation_confidence, (mean(stock.confidence for stock in stocks) if stocks else 0.6), statistics_confidence, agentic_confidence, knowledge_confidence, code_confidence, language_confidence, meta_learning_confidence, moe_confidence, loop_confidence, investing_confidence, payment_confidence, quantum_confidence, physics_confidence, ] ), ), ), 2, ) directives: List[StrategicDirective] = [] north_star_metric = ( diagnostics.recommended_kpis[0] if diagnostics.recommended_kpis else "revenue acceleration" ) for index, score in enumerate(top_scores): channel = score.campaign.channel channel_key = _channel_key(channel) allocation = budget_map.get(channel_key) experiment = experiment_map.get(channel_key) learnings = learning_map.get(channel_key, []) competitors = competitive_map.get(channel_key, []) google_signal = google_ads[index % len(google_ads)] if google_ads else None affiliate_signal = cj_affiliate[index % len(cj_affiliate)] if cj_affiliate else None amazon_signal = amazon[index % len(amazon)] if amazon else None ebay_signal = ebay[index % len(ebay)] if ebay else None paypal_signal = paypal[index % len(paypal)] if paypal else None payment_plan = payments[index % len(payments)] if payments else None influencer_signal = ( influencer_pool[index % len(influencer_pool)] if influencer_pool else None ) investing_plan = investing[index % len(investing)] if investing else None stock_signal = stocks[index % len(stocks)] if stocks else None cognitive_plan = cognitive[index % len(cognitive)] if cognitive else None agentic_plan = agentic[index % len(agentic)] if agentic else None knowledge_plan = knowledge[index % len(knowledge)] if knowledge else None marketing_plan = ( marketing_knowledge[index % len(marketing_knowledge)] if marketing_knowledge else None ) language_plan = language[index % len(language)] if language else None meta_plan = meta_learning[index % len(meta_learning)] if meta_learning else None moe_plan = moe[0] if moe else None loop_plan = agentic_loops[index % len(agentic_loops)] if agentic_loops else None quantum_plan = quantum[index % len(quantum)] if quantum else None physics_plan = ( physics_economics[index % len(physics_economics)] if physics_economics else None ) supporting_channels = [channel] if google_signal: supporting_channels.append("Google Ads") if affiliate_signal: supporting_channels.append("Affiliate") if amazon_signal: supporting_channels.append("Marketplace") if ebay_signal: supporting_channels.append("eBay") if paypal_signal: supporting_channels.append("PayPal") if payment_plan: supporting_channels.append(f"Payments - {payment_plan.provider}") if investing_plan: supporting_channels.append(f"Investing - {investing_plan.provider}") if influencer_signal: supporting_channels.append(f"Influencer - {influencer_signal.platform}") if stock_signal: supporting_channels.append(f"Markets - {stock_signal.symbol}") if automation and automation.tasks: supporting_channels.append("Automation Sprint") if statistics: supporting_channels.append("Statistical Reasoning") if cognitive: supporting_channels.append(f"Cognitive - {cognitive[0].capability}") if agentic: supporting_channels.append(f"Agentic - {agentic[0].system_name}") if knowledge_plan: supporting_channels.append(f"Knowledge - {knowledge_plan.category.upper()}") if marketing_plan: supporting_channels.append(f"Knowledge Sync - {marketing_plan.name}") if language_plan: supporting_channels.append(f"Language - {language_plan.capability}") if meta_plan: supporting_channels.append(f"Meta-Learning - {meta_plan.capability}") if moe_plan: supporting_channels.append("Mixture-of-Experts") if loop_plan: supporting_channels.append("Agentic Loop") if quantum_plan: supporting_channels.append(f"Quantum - {quantum_plan.simulator}") if physics_plan: supporting_channels.append( f"Physics/Economics - {physics_plan.capability}" ) supporting_channels = list(dict.fromkeys(supporting_channels)) action_steps: List[str] = [ ( f"Activate {channel} program targeting {score.campaign.audience} to " f"{score.campaign.objective}." ) ] if experiment: action_steps.append( f"Run Experiment '{experiment.name}' optimizing {experiment.primary_metric}." ) if google_signal: action_steps.append( f"Model spend with Google Ads ROAS {google_signal.roas:.1f}x benchmark." ) if affiliate_signal: commission_pct = affiliate_signal.commission_rate * 100 action_steps.append( f"Layer CJ affiliate offer '{affiliate_signal.program_name}' at {commission_pct:.1f}% commission." ) if amazon_signal: action_steps.append( f"Promote marketplace listing '{amazon_signal.title}' to unlock partner revenue." ) if ebay_signal: action_steps.append( ( f"Launch eBay offer '{ebay_signal.title}' leveraging {ebay_signal.affiliate_rate * 100:.1f}%" " partner share and rapid sell-through." ) ) if paypal_signal: action_steps.append( ( f"Enable PayPal {paypal_signal.checkout_flow} flow with {paypal_signal.payout_schedule} " "payout cadence for dependable paychecks." ) ) if payment_plan: focus = payment_plan.integration_focus[0] if payment_plan.integration_focus else "checkout" action_steps.append( f"Deploy {payment_plan.provider} {focus.lower()} runbook to operationalize payments and follow troubleshooting/codefix guides." ) if investing_plan: asset_summary = ", ".join(investing_plan.asset_classes[:2]) or "key assets" action_steps.append( f"Launch {investing_plan.provider} investing funnel targeting {asset_summary}." ) if stock_signal: action_steps.append( ( f"Align budget pacing with {stock_signal.name} {stock_signal.trend} trend " f"({stock_signal.forecast_return_30d*100:.1f}% 30d outlook)." ) ) if influencer_signal: action_steps.append( ( f"Activate {influencer_signal.collab_format} with {influencer_signal.name} " f"to reach {influencer_signal.estimated_reach:,.0f} engaged prospects." ) ) if automation and automation.tasks: first_task = automation.tasks[0] kpi_focus = first_task.kpis[0] if first_task.kpis else "primary KPI" action_steps.append( f"Kick off automation sprint task '{first_task.activity}' for {kpi_focus}." ) if statistics: top_driver = max( statistics.driver_confidence.items(), key=lambda item: item[1], )[0].split(":", 1)[1] if statistics.driver_confidence else "core drivers" action_steps.append( f"Instrument statistical driver '{top_driver}' to validate {statistics.expected_roi:.2f}x ROI forecast." ) if cognitive_plan: action_steps.append( ( f"Operationalize {cognitive_plan.capability} via {cognitive_plan.frameworks[0]} " "to keep ANI reasoning adaptive." ) ) if agentic_plan: primary_tool = agentic_plan.toolchain[0] if agentic_plan.toolchain else "agent orchestrator" action_steps.append( ( f"Deploy agentic system {agentic_plan.system_name} using {primary_tool} " "to unify tool calling and database flows." ) ) if knowledge_plan: action_steps.append( ( f"Refresh {knowledge_plan.category.upper()} workflow '{knowledge_plan.name}' " "to ground recommendations with up-to-date context." ) ) if marketing_plan: action_steps.append( ( f"Run {marketing_plan.update_frequency} sync for {marketing_plan.name} " "so marketing intelligence stays current." ) ) if language_plan: action_steps.append( ( f"Deploy language workflow '{language_plan.capability}' to keep prompts, chats, and copy aligned with revenue goals." ) ) if moe_plan and moe_plan.activations: top_activation = moe_plan.activations[0] action_steps.append( ( f"Route monetization decisions through {top_activation.name} ({top_activation.provider}) " f"at {top_activation.confidence:.0%} confidence." ) ) if loop_plan: action_steps.append( ( f"Run agentic loop '{loop_plan.name}' to cycle perception→action across revenue systems." ) ) if quantum_plan: top_scenario = quantum_plan.stress_tests[0] if quantum_plan.stress_tests else None if top_scenario: action_steps.append( ( f"Execute quantum stress '{top_scenario.name}' to validate {top_scenario.domain} readiness before scale." ) ) else: action_steps.append( f"Review quantum simulator {quantum_plan.simulator} telemetry prior to ramping spend." ) if physics_plan: action_steps.append( ( f"Activate physics/economics orchestration '{physics_plan.capability}'" " to align complex market mechanics with revenue plays." ) ) if learnings: next_action = ( learnings[0].next_actions[0] if learnings[0].next_actions else "capture fresh experiment data" ) action_steps.append( f"Apply learning '{learnings[0].learning}' with next action '{next_action}'." ) action_steps = list(dict.fromkeys(action_steps)) dependencies: List[str] = [] if allocation: dependencies.append( f"Allocate ${allocation.allocation_usd:,.0f} to {channel} with expected ${allocation.expected_revenue:,.0f} returns" ) else: dependencies.append("Confirm working budget envelope for channel execution") if google_signal and google_signal.source != "simulated": dependencies.append("Validate Google Ads account credentials and conversion tracking") if affiliate_signal and affiliate_signal.source != "cj_catalog": dependencies.append("Sync CJ affiliate program approvals") if automation and automation.tasks: dependencies.append("Reserve automation operators for 3-hour sprint") if ebay_signal and ebay_signal.source != "ebay_catalog": dependencies.append("Confirm eBay developer token and partner campaign tracking") if paypal_signal and paypal_signal.source != "simulated": dependencies.append("Verify PayPal payout configuration and webhook capture") elif paypal_signal: dependencies.append("Export PAYPAL credentials to activate live paychecks") if influencer_signal: est_investment = influencer_signal.projected_cpm * influencer_signal.estimated_reach / 1000.0 dependencies.append( ( f"Fund influencer collaboration (~${est_investment:,.0f}) to secure " f"{influencer_signal.projected_roi:.1f}x projected ROI" ) ) if stock_signal: dependencies.append( ( f"Track {stock_signal.symbol} volatility ({stock_signal.volatility:.2f}) " "to schedule reinvestment windows." ) ) if statistics and statistics.success_probability < 0.7: dependencies.append( "Schedule mid-cycle statistical refresh to bolster success probability" ) if investing_plan: dependencies.append( f"Coordinate compliance review for {investing_plan.provider} investing campaigns" ) if cognitive_plan: dependencies.append( ( f"Deploy {cognitive_plan.capability} ops with {cognitive_plan.tooling[0]} instrumentation" ) ) if agentic_plan: dependencies.append( ( f"Enable {agentic_plan.system_name} access to {agentic_plan.data_interfaces[0] if agentic_plan.data_interfaces else 'core data systems'}" ) ) if knowledge_plan: dependencies.append( f"Maintain {knowledge_plan.category.upper()} store '{knowledge_plan.vector_store}' freshness cadence" ) if language_plan: dependencies.append( f"Provision language stack ({', '.join(language_plan.tooling[:2] or ['NLP tooling'])}) to operationalize {language_plan.capability}" ) if moe_plan and not moe_plan.activations: dependencies.append("Configure MoE router credentials to reach external experts") if loop_plan: dependencies.append("Instrument loop telemetry dashboards for autonomous retrospectives") if quantum_plan: avg_stability = ( mean(test.stability_index for test in quantum_plan.stress_tests) if quantum_plan.stress_tests else 0.5 ) dependencies.append( ( f"Review quantum stress telemetry from {quantum_plan.simulator} " f"(avg stability {avg_stability:.2f}) before campaign scale." ) ) if physics_plan: dependencies.append( ( f"Coordinate physics/economics plan {physics_plan.capability} docs" " to align hardware, compression, and market guards." ) ) risk_mitigations: List[str] = [] for risk in diagnostics.risk_signals[:2]: risk_mitigations.append(f"Mitigate risk: {risk}") for competitor in competitors[:1]: risk_mitigations.append( f"Counter {competitor.competitor} via {competitor.recommended_play}" ) if learnings and learnings[0].losses: risk_mitigations.append( f"Avoid prior loss: {learnings[0].losses[0]}" ) if influencer_signal and influencer_signal.projected_roi < 3.5: risk_mitigations.append( "Negotiate performance clauses with influencer partner to protect ROI" ) if stock_signal and stock_signal.volatility > 0.28: risk_mitigations.append( f"Hedge equity exposure while {stock_signal.symbol} volatility runs {stock_signal.volatility:.2f}" ) if statistics and statistics.volatility_adjustment > 0.25: risk_mitigations.append( "Stage spend releases to account for statistical volatility adjustments" ) if investing_plan and investing_plan.risk_controls: risk_mitigations.append( f"Apply investing guardrail: {investing_plan.risk_controls[0]}" ) if cognitive_plan and cognitive_plan.maturity == "foundational": risk_mitigations.append( f"Invest in advancing {cognitive_plan.capability} maturity for sustained autonomy" ) if agentic_plan and agentic_plan.maturity in {"foundational", "advanced"}: risk_mitigations.append( f"Upskill agentic system {agentic_plan.system_name} workflows for higher autonomy" ) if code_automation and code_automation[0].maturity == "foundational": risk_mitigations.append( f"Harden code automation {code_automation[0].capability} before shipping critical releases" ) if knowledge_plan and knowledge_plan.maturity == "foundational": risk_mitigations.append( f"Invest in elevating {knowledge_plan.category.upper()} workflow '{knowledge_plan.name}' maturity" ) if marketing_plan and marketing_plan.maturity == "foundational": risk_mitigations.append( f"Enable live connectors for {marketing_plan.name} to accelerate knowledge automation" ) if language_plan and language_plan.maturity == "foundational": risk_mitigations.append( f"Upskill language workflow '{language_plan.capability}' for autonomous conversational coverage" ) if moe_plan and not moe_plan.activations: risk_mitigations.append("Map fallbacks for MoE router until experts respond in real time") if loop_plan and loop_plan.maturity != "autonomous": risk_mitigations.append( f"Calibrate agentic loop '{loop_plan.name}' checkpoints before scaling spend" ) if quantum_plan and quantum_plan.stress_tests: worst_stability = min(test.stability_index for test in quantum_plan.stress_tests) if worst_stability < 0.5: risk_mitigations.append( "Instrument fallback flows for domains flagged by quantum stress instability" ) else: risk_mitigations.append( "Monitor quantum telemetry for emerging instability across revenue systems" ) if physics_plan and physics_plan.maturity in {"emerging", "advanced"}: risk_mitigations.append( f"Backtest physics/economics plan {physics_plan.capability} against live market shocks" ) if not risk_mitigations: risk_mitigations.append("Maintain rapid feedback loops across processes") directive_confidence = round( min( 0.99, max( 0.3, mean( [ score.metrics.confidence, min(0.95, max(0.3, score.metrics.expected_roi / 5.0)), learnings[0].confidence if learnings else diagnostics.confidence, ] ), ), ), 2, ) time_horizon = base_horizon + index * 48 supporting_metrics_set: set[str] = {north_star_metric} if experiment: supporting_metrics_set.add(experiment.primary_metric) supporting_metrics_set.update(experiment.secondary_metrics) if influencer_signal: supporting_metrics_set.add("influencer reach") if stock_signal: supporting_metrics_set.add("market return correlation") if statistics: supporting_metrics_set.add("statistical success probability") if investing_plan and investing_plan.kpis: supporting_metrics_set.add(investing_plan.kpis[0]) if cognitive_plan: supporting_metrics_set.add("cognitive readiness index") if agentic_plan: supporting_metrics_set.add(agentic_plan.metrics[0] if agentic_plan.metrics else "agentic readiness") if language_plan and language_plan.evaluation_metrics: supporting_metrics_set.add(language_plan.evaluation_metrics[0]) if meta_plan and meta_plan.metrics: supporting_metrics_set.add(meta_plan.metrics[0]) if code_automation: supporting_metrics_set.add( code_automation[0].metrics[0] if code_automation[0].metrics else "automation stability" ) if marketing_plan and marketing_plan.metrics: supporting_metrics_set.add(marketing_plan.metrics[0]) if quantum_plan: supporting_metrics_set.add("quantum stability") if physics_plan: supporting_metrics_set.add("physics_economics_readiness") directive = StrategicDirective( name=f"Directive {index + 1}: {score.campaign.channel.title()} Surge", priority=index + 1, objective=score.campaign.objective, primary_channel=channel, supporting_channels=supporting_channels, investment_usd=round(allocation.allocation_usd, 2) if allocation else None, expected_revenue_usd=round(allocation.expected_revenue, 2) if allocation else None, confidence=directive_confidence, time_horizon_hours=time_horizon, north_star_metric=north_star_metric, supporting_metrics=sorted(supporting_metrics_set), action_steps=action_steps, dependencies=dependencies, risk_mitigations=risk_mitigations, inspiration=" | ".join( filter( None, [ experiment.name if experiment else None, learnings[0].play_name if learnings else None, competitors[0].competitor if competitors else None, rag_channels.get(index), ], ) ) or "Campaign intelligence fusion", ) directives.append(directive) summary = ( f"Synthesize {len(directives)} directives to accelerate {diagnostics.stage} outcomes " f"for {diagnostics.focus}." ) notes_candidates = [] if excellence: notes_candidates.append(excellence.headline) if budget: notes_candidates.append(budget.summary) if automation: notes_candidates.append(automation.summary) if influencer_pool: notes_candidates.append( ( f"Influencer plan anchored by {influencer_pool[0].name} " f"({influencer_pool[0].projected_roi:.1f}x ROI outlook)" ) ) if stocks: top_stock = stocks[0] notes_candidates.append( ( f"Market outlook: {top_stock.name} {top_stock.trend} with " f"{top_stock.forecast_return_90d*100:.0f}% 90d upside" ) ) if statistics: notes_candidates.append(statistics.narrative) if cognitive: notes_candidates.append( f"Cognitive architecture anchored by {cognitive[0].capability} ({cognitive[0].maturity})." ) if language: notes_candidates.append( f"Language workflow {language[0].capability} ({language[0].maturity}) powering conversational loops." ) if meta_learning: notes_candidates.append( f"Meta-learning {meta_learning[0].capability} ({meta_learning[0].maturity}) keeps experimentation adaptive." ) if code_automation: notes_candidates.append( f"Code automation {code_automation[0].capability} ({code_automation[0].maturity}) safeguards release velocity." ) if payments: notes_candidates.append( f"Payments orchestrated via {payments[0].provider} ({payments[0].maturity})." ) if investing: notes_candidates.append( f"Investing runway anchored by {investing[0].provider} ({investing[0].maturity})." ) if agentic: notes_candidates.append( f"Agentic system {agentic[0].system_name} ({agentic[0].maturity}) orchestrates tool stack." ) if quantum: notes_candidates.append( f"Quantum simulator {quantum[0].simulator} covers {len(quantum[0].stress_tests)} stress scenarios." ) if physics_economics: notes_candidates.append( f"Physics/economics plan {physics_economics[0].capability} ({physics_economics[0].maturity}) guiding complex mechanics." ) notes = " | ".join(notes_candidates) if notes_candidates else None return ExecutivePlaybook( headline= f"{diagnostics.stage.title()} momentum for {diagnostics.focus} with autonomous orchestration", focus=diagnostics.focus, urgency=diagnostics.urgency, north_star_metric=north_star_metric, total_budget_usd=round(budget.total_budget, 2) if budget else None, expected_revenue_usd=round(budget.expected_revenue, 2) if budget else None, confidence=overall_confidence, directives=directives, summary=summary, notes=notes, ) @lru_cache(maxsize=1) def default_marketing_engine() -> MarketingIntelligenceEngine: return MarketingIntelligenceEngine.from_file()