Janady13/AI_Brain / services/seeder_api/intelligence.py

1"""Marketing intelligence engine that ranks and scores enumerated campaigns.""" 2 3from __future__ import annotations 4 5import base64 6import importlib 7import json 8import math 9import os 10import re 11from dataclasses import dataclass 12from functools import lru_cache 13from statistics import mean 14from typing import Any, Dict, List, Sequence, Tuple 15 16try: # pragma: no cover - optional dependency that may not be installed in tests 17 from google.ads.googleads.client import GoogleAdsClient # type: ignore 18except ImportError: # pragma: no cover - gracefully handled when SDK missing 19 GoogleAdsClient = None # type: ignore 20 21try: # pragma: no cover - optional dependency that may not be installed in tests 22 from google.oauth2 import service_account # type: ignore 23except ImportError: # pragma: no cover - gracefully handled when SDK missing 24 service_account = None # type: ignore 25 26try: # pragma: no cover - optional dependency that may not be installed in tests 27 import stripe # type: ignore 28except ImportError: # pragma: no cover - gracefully handled when SDK missing 29 stripe = None # type: ignore 30 31try: # pragma: no cover - optional dependency that may not be installed in tests 32 from paypalcheckoutsdk.core import ( # type: ignore 33 PayPalHttpClient, 34 LiveEnvironment, 35 SandboxEnvironment, 36 ) 37except ImportError: # pragma: no cover - gracefully handled when SDK missing 38 PayPalHttpClient = None # type: ignore 39 LiveEnvironment = None # type: ignore 40 SandboxEnvironment = None # type: ignore 41 42try: # pragma: no cover - optional dependency for AWS integrations 43 import boto3 # type: ignore 44except ImportError: # pragma: no cover - gracefully handled when SDK missing 45 boto3 = None # type: ignore 46 47try: # pragma: no cover - optional dependency for Postgres persistence 48 import psycopg2 # type: ignore 49 from psycopg2.extras import Json # type: ignore 50except Exception: # pragma: no cover - handle environments without psycopg2 51 psycopg2 = None # type: ignore 52 Json = None # type: ignore 53 54 55MARKETING_ENUMERATION = os.path.join( 56 os.path.dirname(__file__), "..", "..", "daemon", "enumerations", "marketing_500.txt" 57) 58MARKETING_BENCHMARKS = os.path.join( 59 os.path.dirname(__file__), "..", "..", "daemon", "datasets", "marketing_benchmarks.json" 60) 61CJ_OFFERS_DATASET = os.path.join( 62 os.path.dirname(__file__), "..", "..", "daemon", "datasets", "cj_offers.json" 63) 64AMAZON_MARKETPLACE_DATASET = os.path.join( 65 os.path.dirname(__file__), "..", "..", "daemon", "datasets", "amazon_marketplace.json" 66) 67EBAY_MARKETPLACE_DATASET = os.path.join( 68 os.path.dirname(__file__), "..", "..", "daemon", "datasets", "ebay_marketplace.json" 69) 70AWS_GROWTH_PLAYBOOKS = os.path.join( 71 os.path.dirname(__file__), "..", "..", "daemon", "datasets", "aws_growth_playbooks.json" 72) 73CHANNEL_INVESTMENTS_DATASET = os.path.join( 74 os.path.dirname(__file__), 75 "..", 76 "..", 77 "daemon", 78 "datasets", 79 "channel_investments.json", 80) 81 82AUTOMATION_SPRINTS_DATASET = os.path.join( 83 os.path.dirname(__file__), 84 "..", 85 "..", 86 "daemon", 87 "datasets", 88 "automation_sprints.json", 89) 90 91PERFORMANCE_MEMORY_DATASET = os.path.join( 92 os.path.dirname(__file__), 93 "..", 94 "..", 95 "daemon", 96 "datasets", 97 "performance_memory.json", 98) 99 100COMPETITIVE_LANDSCAPE_DATASET = os.path.join( 101 os.path.dirname(__file__), 102 "..", 103 "..", 104 "daemon", 105 "datasets", 106 "competitive_landscape.json", 107) 108 109INFLUENCER_PARTNERS_DATASET = os.path.join( 110 os.path.dirname(__file__), 111 "..", 112 "..", 113 "daemon", 114 "datasets", 115 "influencer_partners.json", 116) 117 118STOCK_MARKETS_DATASET = os.path.join( 119 os.path.dirname(__file__), 120 "..", 121 "..", 122 "daemon", 123 "datasets", 124 "global_stock_markets.json", 125) 126 127STATISTICAL_PRIORS_DATASET = os.path.join( 128 os.path.dirname(__file__), 129 "..", 130 "..", 131 "daemon", 132 "datasets", 133 "statistical_reasoning.json", 134) 135 136COGNITIVE_CAPABILITIES_DATASET = os.path.join( 137 os.path.dirname(__file__), 138 "..", 139 "..", 140 "daemon", 141 "datasets", 142 "cognitive_capabilities.json", 143) 144 145AGENTIC_SYSTEMS_DATASET = os.path.join( 146 os.path.dirname(__file__), 147 "..", 148 "..", 149 "daemon", 150 "datasets", 151 "agentic_systems.json", 152) 153 154requests_spec = importlib.util.find_spec("requests") 155requests = importlib.import_module("requests") if requests_spec else None 156 157 158def _load_lines(path: str) -> List[str]: 159 with open(path, "r", encoding="utf-8") as handle: 160 return [line.strip() for line in handle if line.strip()] 161 162 163@dataclass(slots=True) 164class MarketingCampaign: 165 """Structured representation for one marketing campaign idea.""" 166 167 raw: str 168 channel: str 169 audience: str 170 offer: str 171 objective: str 172 173 @classmethod 174 def parse(cls, raw: str) -> "MarketingCampaign": 175 text = raw.strip().rstrip(".") 176 if not text.lower().startswith("campaign:"): 177 raise ValueError(f"Unsupported campaign format: {raw}") 178 179 body = text[len("Campaign:") :].strip() 180 181 try: 182 channel, remainder = body.split("targeting", 1) 183 audience, remainder = remainder.split("leveraging", 1) 184 offer, objective = remainder.split(" to ", 1) 185 except ValueError as exc: # pragma: no cover - enumeration should be well formed 186 raise ValueError(f"Malformed campaign sentence: {raw}") from exc 187 188 return cls( 189 raw=raw, 190 channel=channel.strip(), 191 audience=audience.strip(), 192 offer=offer.strip(), 193 objective=objective.strip().rstrip("."), 194 ) 195 196 197@dataclass(slots=True) 198class CampaignMetrics: 199 """Statistical expectations for a marketing campaign.""" 200 201 expected_roi: float 202 expected_ctr: float 203 confidence: float 204 drivers: Dict[str, float] 205 206 207@dataclass(slots=True) 208class StatisticalPrior: 209 """Historical statistical benchmark for a marketing dimension.""" 210 211 feature: str 212 feature_type: str 213 mean_roi: float 214 std_roi: float 215 mean_ctr: float 216 std_ctr: float 217 sample_size: int 218 weight: float 219 baseline_budget: float 220 notes: str 221 222 223@dataclass(slots=True) 224class CognitiveCapability: 225 """Represents a cognitive pillar the ANI can operationalize.""" 226 227 name: str 228 pillar: str 229 description: str 230 signal_keywords: List[str] 231 prompt_strategies: List[str] 232 frameworks: List[str] 233 tooling: List[str] 234 metrics: List[str] 235 knowledge_assets: List[str] 236 play: str 237 238 @classmethod 239 def from_dict(cls, payload: Dict[str, Any]) -> "CognitiveCapability": 240 return cls( 241 name=payload["name"], 242 pillar=payload["pillar"], 243 description=payload["description"], 244 signal_keywords=[keyword.lower() for keyword in payload["signal_keywords"]], 245 prompt_strategies=payload["prompt_strategies"], 246 frameworks=payload["frameworks"], 247 tooling=payload["tooling"], 248 metrics=payload["metrics"], 249 knowledge_assets=payload["knowledge_assets"], 250 play=payload["play"], 251 ) 252 253 254@dataclass(slots=True) 255class CognitiveCapabilityPlan: 256 """Cognitive orchestration outcome derived from prompt and campaign context.""" 257 258 capability: str 259 pillar: str 260 maturity: str 261 description: str 262 reasoning: str 263 prompt_strategy: str 264 frameworks: List[str] 265 tooling: List[str] 266 metrics: List[str] 267 knowledge_assets: List[str] 268 orchestration_steps: List[str] 269 source: str 270 271 272@dataclass(slots=True) 273class AgenticCapability: 274 """Represents an agentic system capability that can be activated.""" 275 276 name: str 277 domain: str 278 description: str 279 controller: str 280 trigger_keywords: List[str] 281 frameworks: List[str] 282 toolchain: List[str] 283 data_interfaces: List[str] 284 knowledge_sources: List[str] 285 revenue_programs: List[str] 286 risk_controls: List[str] 287 maturity_signals: List[str] 288 base_maturity: str 289 metrics: List[str] 290 291 @classmethod 292 def from_dict(cls, payload: Dict[str, Any]) -> "AgenticCapability": 293 return cls( 294 name=payload["name"], 295 domain=payload["domain"], 296 description=payload["description"], 297 controller=payload.get("controller", "Agent controller"), 298 trigger_keywords=[keyword.lower() for keyword in payload.get("trigger_keywords", [])], 299 frameworks=payload.get("frameworks", []), 300 toolchain=payload.get("toolchain", []), 301 data_interfaces=payload.get("data_interfaces", []), 302 knowledge_sources=payload.get("knowledge_sources", []), 303 revenue_programs=payload.get("revenue_programs", []), 304 risk_controls=payload.get("risk_controls", []), 305 maturity_signals=payload.get("maturity_signals", []), 306 base_maturity=payload.get("base_maturity", "foundational"), 307 metrics=payload.get("metrics", []), 308 ) 309 310 311@dataclass(slots=True) 312class AgenticSystemPlan: 313 """Structured plan describing how the ANI activates agentic systems.""" 314 315 system_name: str 316 domain: str 317 controller: str 318 maturity: str 319 reasoning: str 320 orchestration_steps: List[str] 321 toolchain: List[str] 322 data_interfaces: List[str] 323 knowledge_sources: List[str] 324 revenue_programs: List[str] 325 risk_controls: List[str] 326 metrics: List[str] 327 source: str 328 notes: str | None = None 329 330 331@dataclass(slots=True) 332class StatisticalEvidence: 333 """Posterior insight after blending priors with live campaign metrics.""" 334 335 feature: str 336 feature_type: str 337 posterior_roi: float 338 posterior_ctr: float 339 confidence: float 340 lift_vs_benchmark: float 341 sample_size: int 342 weight: float 343 rationale: str 344 345 346@dataclass(slots=True) 347class StatisticalReasoningSummary: 348 """Aggregated statistical reasoning powering marketing ANI.""" 349 350 expected_roi: float 351 expected_revenue: float 352 confidence: float 353 success_probability: float 354 volatility_adjustment: float 355 scenario_spread: Dict[str, float] 356 driver_confidence: Dict[str, float] 357 evidence: List[StatisticalEvidence] 358 narrative: str 359 source: str = "statistical_reasoning" 360 notes: str | None = None 361 362 363@dataclass(slots=True) 364class BudgetProfile: 365 """Baseline investment expectations for a marketing channel.""" 366 367 channel: str 368 keywords: List[str] 369 baseline_cpc: float 370 baseline_cpl: float 371 avg_conversion_rate: float 372 avg_roi: float 373 recommended_monthly_spend: float 374 375 376@dataclass(slots=True) 377class BudgetAllocation: 378 """Recommended allocation for a marketing campaign channel.""" 379 380 channel: str 381 audience_focus: str 382 offer: str 383 objective: str 384 allocation_usd: float 385 allocation_weight: float 386 expected_leads: float 387 expected_revenue: float 388 rationale: str 389 source: str 390 notes: str | None = None 391 392 393@dataclass(slots=True) 394class BudgetPlan: 395 """Portfolio-level budget recommendation summary.""" 396 397 total_budget: float 398 allocations: List[BudgetAllocation] 399 expected_revenue: float 400 expected_roi: float 401 payback_period_weeks: float 402 summary: str 403 404 405@dataclass(slots=True) 406class AutomationTask: 407 """Time-boxed task within an automation sprint.""" 408 409 start_minute: int 410 end_minute: int 411 activity: str 412 outcome: str 413 tooling: List[str] 414 kpis: List[str] 415 automation_level: str 416 source: str 417 418 419@dataclass(slots=True) 420class AutomationSprint: 421 """Three-hour automation run composed of sequential tasks.""" 422 423 total_minutes: int 424 stage: str 425 focus: str 426 summary: str 427 tasks: List[AutomationTask] 428 notes: str | None = None 429 source: str = "automation_playbooks" 430 431 432@dataclass(slots=True) 433class PerformanceMemoryEntry: 434 """Historical marketing learning captured from closed-loop experiments.""" 435 436 play_name: str 437 stage: str 438 audience_focus: str 439 channel: str 440 tags: List[str] 441 summary: str 442 wins: List[str] 443 losses: List[str] 444 next_actions: List[str] 445 kpi_focus: List[str] 446 confidence: float 447 source: str = "performance_memory" 448 notes: str | None = None 449 450 451@dataclass(slots=True) 452class LearningLoopRecommendation: 453 """Actionable recommendation generated from the learning memory.""" 454 455 play_name: str 456 stage: str 457 audience_focus: str 458 channel: str 459 learning: str 460 wins: List[str] 461 losses: List[str] 462 next_actions: List[str] 463 kpi_focus: List[str] 464 confidence: float 465 source: str 466 notes: str | None = None 467 468 469@dataclass(slots=True) 470class ExcellenceSummary: 471 """Composite readiness score describing how robust the ANI recommendations are.""" 472 473 excellence_index: float 474 maturity_level: str 475 headline: str 476 strengths: List[str] 477 watchouts: List[str] 478 process_coverage: Dict[str, float] 479 notes: str | None = None 480 481 482@dataclass(slots=True) 483class StrategicDirective: 484 """Directive that fuses multiple process signals into a prioritized motion.""" 485 486 name: str 487 priority: int 488 objective: str 489 primary_channel: str 490 supporting_channels: List[str] 491 investment_usd: float | None 492 expected_revenue_usd: float | None 493 confidence: float 494 time_horizon_hours: int 495 north_star_metric: str 496 supporting_metrics: List[str] 497 action_steps: List[str] 498 dependencies: List[str] 499 risk_mitigations: List[str] 500 inspiration: str 501 source: str = "ani_orchestration" 502 notes: str | None = None 503 504 505@dataclass(slots=True) 506class ExecutivePlaybook: 507 """Cross-process action plan that guides autonomous execution.""" 508 509 headline: str 510 focus: str 511 urgency: str 512 north_star_metric: str 513 total_budget_usd: float | None 514 expected_revenue_usd: float | None 515 confidence: float 516 directives: List[StrategicDirective] 517 summary: str 518 notes: str | None = None 519 520 521@dataclass(slots=True) 522class ScenarioProjection: 523 """Scenario-level forecast for autonomous marketing execution.""" 524 525 scenario: str 526 expected_revenue: float 527 expected_roi: float 528 probability: float 529 confidence: float 530 accelerators: List[str] 531 stabilizers: List[str] 532 watchouts: List[str] 533 source: str = "scenario_planner" 534 notes: str | None = None 535 536 537@dataclass(slots=True) 538class ScenarioAnalysis: 539 """Aggregated scenario planning summary.""" 540 541 headline: str 542 summary: str 543 projections: List[ScenarioProjection] 544 confidence: float 545 accelerators: List[str] 546 stabilizers: List[str] 547 source: str = "scenario_planner" 548 notes: str | None = None 549 550 551class ScenarioPlanningProcess: 552 """Synthesizes optimistic, base, and conservative revenue scenarios.""" 553 554 def __init__(self, volatility_floor: float = 0.15) -> None: 555 self._volatility_floor = volatility_floor 556 557 def _estimate_budget( 558 self, diagnostics: PromptDiagnostics, campaigns: Sequence[CampaignScore] 559 ) -> float: 560 stage_baseline = { 561 "awareness": 18000.0, 562 "consideration": 22000.0, 563 "conversion": 28000.0, 564 "retention": 20000.0, 565 } 566 base = stage_baseline.get(diagnostics.stage, 22000.0) 567 focus = diagnostics.focus.lower() 568 if "enterprise" in focus: 569 base *= 1.45 570 elif "startup" in focus: 571 base *= 0.9 572 elif "consumer" in focus: 573 base *= 0.8 574 575 if diagnostics.urgency == "urgent": 576 base *= 1.25 577 elif diagnostics.urgency == "time-sensitive": 578 base *= 1.1 579 580 avg_roi = mean(score.metrics.expected_roi for score in campaigns) 581 base *= max(0.8, min(1.6, avg_roi / 2.0)) 582 583 return round(max(7500.0, base), 2) 584 585 @staticmethod 586 def _signal_accelerators( 587 google_ads: Sequence[GoogleCampaignPerformance] | None, 588 cj_affiliate: Sequence[CJAffiliateRecommendation] | None, 589 amazon: Sequence[AmazonProductRecommendation] | None, 590 ebay: Sequence[EbayRevenuePlay] | None, 591 aws: Sequence[AWSRevenueRecommendation] | None, 592 stripe: Sequence[StripeMonetizationPlan] | None, 593 paypal: Sequence[PayPalMonetizationPlan] | None, 594 influencers: Sequence[InfluencerRecommendation] | None, 595 stocks: Sequence[StockMarketForecast] | None, 596 statistics: StatisticalReasoningSummary | None, 597 cognitive: Sequence[CognitiveCapabilityPlan] | None, 598 agentic: Sequence[AgenticSystemPlan] | None, 599 ) -> List[str]: 600 accelerators: List[str] = [] 601 if google_ads: 602 roas = mean(perf.roas for perf in google_ads) 603 if roas >= 3.5: 604 accelerators.append(f"Google Ads ROAS averaging {roas:.2f}x supports paid scale") 605 if cj_affiliate: 606 revenue = mean(rec.projected_monthly_revenue for rec in cj_affiliate) 607 if revenue > 15000: 608 accelerators.append( 609 f"CJ affiliate programs forecast ${revenue:,.0f} monthly expansion" 610 ) 611 if amazon: 612 conversion = mean(rec.rating for rec in amazon) 613 if conversion >= 4.2: 614 accelerators.append("Amazon marketplace sentiment reinforces conversion confidence") 615 if ebay: 616 margin = mean(play.projected_profit_margin for play in ebay) 617 velocity = mean(play.sold_per_month for play in ebay) 618 if margin >= 0.25 and velocity >= 100: 619 accelerators.append( 620 "eBay partner inventory blends high-margin payouts with rapid sell-through" 621 ) 622 if aws: 623 roi = mean(rec.projected_roi for rec in aws) 624 if roi >= 2.5: 625 accelerators.append("AWS growth plays carry multi-service ROI leverage") 626 if stripe: 627 price = mean(plan.price_usd for plan in stripe) 628 if price >= 99: 629 accelerators.append("Stripe monetization supports premium pricing uplift") 630 if paypal: 631 payout_focus = {plan.payout_schedule for plan in paypal} 632 if "daily" in payout_focus: 633 accelerators.append("PayPal payouts unlocked daily revenue capture velocity") 634 elif paypal: 635 accelerators.append("PayPal commerce streamlines cashflow-ready offers") 636 if influencers: 637 avg_roi = mean(rec.projected_roi for rec in influencers) 638 avg_reach = mean(rec.estimated_reach for rec in influencers) 639 if avg_roi >= 4.0: 640 accelerators.append( 641 f"Influencer partnerships forecast {avg_roi:.1f}x ROI with amplified reach" 642 ) 643 elif avg_reach >= 5000: 644 accelerators.append( 645 f"Influencer partners extend campaigns to {avg_reach:,.0f}+ engaged prospects" 646 ) 647 if stocks: 648 avg_return = mean(forecast.forecast_return_90d for forecast in stocks) 649 avg_conf = mean(forecast.confidence for forecast in stocks) 650 if avg_return > 0.12 and avg_conf >= 0.55: 651 accelerators.append( 652 f"Equity momentum projects {avg_return*100:.0f}% upside supporting budget expansion" 653 ) 654 elif avg_conf >= 0.6: 655 accelerators.append("Stock market baselines stabilizing capital access for growth plays") 656 if statistics and statistics.success_probability >= 0.75: 657 accelerators.append( 658 f"Statistical reasoning signals {statistics.success_probability:.0%} success with {statistics.expected_roi:.2f}x ROI" 659 ) 660 if cognitive: 661 top_capability = cognitive[0] 662 accelerators.append( 663 f"Cognitive architecture {top_capability.maturity} on {top_capability.capability} keeps orchestration sharp" 664 ) 665 if agentic: 666 system = agentic[0] 667 if system.maturity in {"elite", "autonomous"}: 668 accelerators.append( 669 f"Agentic system {system.system_name} operating at {system.maturity} maturity automates multi-tool execution" 670 ) 671 else: 672 accelerators.append( 673 f"Agentic system {system.system_name} activates {system.toolchain[0] if system.toolchain else 'orchestration'} for rapid deployment" 674 ) 675 return accelerators 676 677 @staticmethod 678 def _stabilizers( 679 diagnostics: PromptDiagnostics, 680 budget: BudgetPlan | None, 681 statistics: StatisticalReasoningSummary | None, 682 cognitive: Sequence[CognitiveCapabilityPlan] | None, 683 agentic: Sequence[AgenticSystemPlan] | None, 684 ) -> List[str]: 685 stabilizers: List[str] = [] 686 if diagnostics.recommended_kpis: 687 stabilizers.append(f"Instrument KPI guardrails around {diagnostics.recommended_kpis[0]}") 688 if budget and budget.allocations: 689 top_channel = budget.allocations[0].channel 690 stabilizers.append(f"Phase spend across {top_channel} with weekly optimization cycles") 691 if statistics: 692 stabilizers.append( 693 f"Recompute statistical priors weekly to maintain {statistics.success_probability:.0%} success guardrails" 694 ) 695 if cognitive: 696 stabilizers.append( 697 f"Audit cognitive capability '{cognitive[0].capability}' to preserve prompt and reasoning quality" 698 ) 699 if agentic: 700 stabilizers.append( 701 f"Schedule agentic system '{agentic[0].system_name}' health checks to keep tool calling compliant" 702 ) 703 return stabilizers 704 705 def analyze( 706 self, 707 prompt: str, 708 diagnostics: PromptDiagnostics, 709 campaigns: Sequence[CampaignScore], 710 *, 711 budget: BudgetPlan | None, 712 google_ads: Sequence[GoogleCampaignPerformance] | None, 713 cj_affiliate: Sequence[CJAffiliateRecommendation] | None, 714 amazon: Sequence[AmazonProductRecommendation] | None, 715 ebay: Sequence[EbayRevenuePlay] | None, 716 aws: Sequence[AWSRevenueRecommendation] | None, 717 stripe: Sequence[StripeMonetizationPlan] | None, 718 paypal: Sequence[PayPalMonetizationPlan] | None, 719 crypto: Sequence[CryptoMarketSnapshot] | None, 720 influencers: Sequence[InfluencerRecommendation] | None, 721 stocks: Sequence[StockMarketForecast] | None, 722 statistics: StatisticalReasoningSummary | None, 723 cognitive: Sequence[CognitiveCapabilityPlan] | None, 724 agentic: Sequence[AgenticSystemPlan] | None, 725 limit: int = 3, 726 ) -> ScenarioAnalysis: 727 if not campaigns: 728 raise ValueError("Scenario planning requires campaign scores") 729 730 top = list(campaigns[: max(1, limit)]) 731 agentic = list(agentic or []) 732 avg_roi = mean(score.metrics.expected_roi for score in top) 733 avg_confidence = mean(score.metrics.confidence for score in top) 734 avg_ctr = mean(score.metrics.expected_ctr for score in top) 735 736 base_budget = ( 737 budget.total_budget if budget and budget.total_budget > 0 else self._estimate_budget(diagnostics, top) 738 ) 739 base_revenue = ( 740 budget.expected_revenue 741 if budget and budget.expected_revenue > 0 742 else round(base_budget * max(1.1, avg_roi), 2) 743 ) 744 base_roi = round(base_revenue / base_budget, 2) if base_budget else round(avg_roi, 2) 745 746 if statistics: 747 inferred_budget = ( 748 statistics.expected_revenue / max(statistics.expected_roi, 0.1) 749 if statistics.expected_roi > 0 750 else base_budget 751 ) 752 base_budget = max(base_budget, inferred_budget) 753 base_revenue = max(base_revenue, statistics.expected_revenue) 754 base_roi = max(base_roi, statistics.expected_roi) 755 756 volatility = self._volatility_floor 757 if diagnostics.urgency == "urgent": 758 volatility += 0.1 759 elif diagnostics.urgency == "time-sensitive": 760 volatility += 0.05 761 if diagnostics.risk_signals: 762 volatility += 0.08 763 if crypto and any(snapshot.change_24h and abs(snapshot.change_24h) > 5 for snapshot in crypto): 764 volatility += 0.05 765 if stocks and any(forecast.volatility > 0.3 for forecast in stocks): 766 volatility += 0.04 767 elif stocks and mean(forecast.volatility for forecast in stocks) < 0.12: 768 volatility -= 0.02 769 if statistics: 770 volatility = max(volatility, statistics.volatility_adjustment) 771 volatility = min(0.5, max(self._volatility_floor, volatility)) 772 773 optimistic_multiplier = 1 + volatility 774 conservative_multiplier = max(0.45, 1 - volatility) 775 776 optimistic_revenue = round(base_revenue * optimistic_multiplier, 2) 777 conservative_revenue = round(base_revenue * conservative_multiplier, 2) 778 779 optimistic_roi = round(base_roi * (1 + volatility / 1.5), 2) 780 conservative_roi = round(base_roi * (1 - volatility / 1.2), 2) 781 782 optimistic_conf = min(0.99, avg_confidence + (0.12 - volatility / 2)) 783 base_conf = min(0.95, max(0.25, avg_confidence)) 784 conservative_conf = max(0.2, avg_confidence - volatility / 2) 785 786 optimistic_prob = max(0.18, 0.32 - (volatility - self._volatility_floor) * 0.6) 787 conservative_prob = max(0.22, 0.28 + (volatility - self._volatility_floor) * 0.5) 788 base_prob = max(0.3, 1.0 - optimistic_prob - conservative_prob) 789 if statistics: 790 blended_base = statistics.success_probability * 0.6 + base_prob * 0.4 791 base_prob = min(0.85, max(0.25, blended_base)) 792 residual = max(0.15, 1.0 - base_prob) 793 share_denominator = max(optimistic_prob + conservative_prob, 0.01) 794 optimistic_share = optimistic_prob / share_denominator 795 optimistic_prob = min(0.6, residual * optimistic_share) 796 conservative_prob = max(0.1, residual - optimistic_prob) 797 total = optimistic_prob + base_prob + conservative_prob 798 optimistic_prob = round(max(0.05, optimistic_prob / total), 2) 799 base_prob = round(max(0.1, base_prob / total), 2) 800 conservative_prob = round(max(0.05, conservative_prob / total), 2) 801 802 accelerators = self._signal_accelerators( 803 google_ads, 804 cj_affiliate, 805 amazon, 806 ebay, 807 aws, 808 stripe, 809 paypal, 810 influencers, 811 stocks, 812 statistics, 813 cognitive, 814 agentic, 815 ) 816 stabilizers = self._stabilizers(diagnostics, budget, statistics, cognitive, agentic) 817 818 if avg_ctr > 0.6: 819 accelerators.append("Benchmark CTR momentum indicates creative resonance") 820 if len(accelerators) < 2: 821 accelerators.append("Cross-process enrichment unlocks multi-channel leverage") 822 if not stabilizers: 823 stabilizers.append("Institute weekly analytics reviews to course-correct") 824 825 risk_watchouts = diagnostics.risk_signals[:2] 826 if not risk_watchouts: 827 risk_watchouts = ["Monitor acquisition efficiency as spend scales"] 828 829 projections = [ 830 ScenarioProjection( 831 scenario="optimistic", 832 expected_revenue=optimistic_revenue, 833 expected_roi=optimistic_roi, 834 probability=optimistic_prob, 835 confidence=round(optimistic_conf, 2), 836 accelerators=accelerators, 837 stabilizers=stabilizers, 838 watchouts=["Ensure fulfillment capacity scales with demand"], 839 ), 840 ScenarioProjection( 841 scenario="base", 842 expected_revenue=round(base_revenue, 2), 843 expected_roi=round(base_roi, 2), 844 probability=base_prob, 845 confidence=round(base_conf, 2), 846 accelerators=accelerators, 847 stabilizers=stabilizers, 848 watchouts=risk_watchouts, 849 ), 850 ScenarioProjection( 851 scenario="conservative", 852 expected_revenue=conservative_revenue, 853 expected_roi=round(conservative_roi, 2), 854 probability=round(conservative_prob, 2), 855 confidence=round(conservative_conf, 2), 856 accelerators=accelerators, 857 stabilizers=stabilizers, 858 watchouts=["Reserve contingency budget to reallocate into proven channels"], 859 ), 860 ] 861 862 primary_channels = {score.campaign.channel for score in top[:2]} 863 summary = ( 864 f"Model optimistic, base, and conservative revenue trajectories for {diagnostics.stage} " 865 f"outcomes targeting {diagnostics.focus}." 866 ) 867 headline = ( 868 f"Scenario coverage for {', '.join(sorted(primary_channels))} anchored in {diagnostics.focus}." 869 ) 870 871 notes = ( 872 f"Volatility factor {volatility:.2f} derives from urgency={diagnostics.urgency} and risk signals" 873 ) 874 if statistics: 875 notes += ( 876 f" | Statistical reasoning forecasts {statistics.expected_roi:.2f}x ROI with " 877 f"{statistics.success_probability:.0%} success probability" 878 ) 879 if agentic: 880 notes += ( 881 f" | Agentic system {agentic[0].system_name} ({agentic[0].maturity}) orchestrates tool calling" 882 ) 883 884 return ScenarioAnalysis( 885 headline=headline, 886 summary=summary, 887 projections=projections, 888 confidence=round(base_conf, 2), 889 accelerators=accelerators, 890 stabilizers=stabilizers, 891 notes=notes, 892 ) 893 894 895class PerformanceMemory: 896 """Loads and scores historical marketing learnings.""" 897 898 def __init__(self, entries: Sequence[PerformanceMemoryEntry]): 899 if not entries: 900 raise ValueError("PerformanceMemory requires at least one entry") 901 self._entries = list(entries) 902 903 @classmethod 904 def from_file(cls, path: str = PERFORMANCE_MEMORY_DATASET) -> "PerformanceMemory": 905 with open(path, "r", encoding="utf-8") as handle: 906 payload = json.load(handle) 907 908 entries = [] 909 for record in payload: 910 entries.append( 911 PerformanceMemoryEntry( 912 play_name=record["play_name"], 913 stage=record["stage"], 914 audience_focus=record["audience_focus"], 915 channel=record["channel"], 916 tags=[tag.lower() for tag in record.get("tags", [])], 917 summary=record["summary"], 918 wins=record.get("wins", []), 919 losses=record.get("losses", []), 920 next_actions=record.get("next_actions", []), 921 kpi_focus=[str(kpi) for kpi in record.get("kpi_focus", [])], 922 confidence=float(record.get("confidence", 0.5)), 923 source=record.get("source", "performance_memory"), 924 notes=record.get("notes"), 925 ) 926 ) 927 928 return cls(entries) 929 930 @property 931 def entries(self) -> List[PerformanceMemoryEntry]: 932 return list(self._entries) 933 934 935@dataclass(slots=True) 936class CompetitiveLandscapeEntry: 937 """Competitive signal captured from market monitoring.""" 938 939 competitor: str 940 channel: str 941 summary: str 942 strengths: List[str] 943 weaknesses: List[str] 944 recommended_play: str 945 share_of_voice: float 946 signal_intensity: float 947 keywords: List[str] 948 maturity: str 949 source: str 950 notes: str | None = None 951 952 953@dataclass(slots=True) 954class CompetitiveSignal: 955 """Recommendation derived from the competitive landscape dataset.""" 956 957 competitor: str 958 channel: str 959 summary: str 960 strengths: List[str] 961 weaknesses: List[str] 962 recommended_play: str 963 share_of_voice: float 964 signal_intensity: float 965 threat_level: str 966 focus_alignment: float 967 source: str 968 notes: str | None = None 969 970 971class CompetitiveLandscape: 972 """Loads competitive marketing signals from curated monitoring.""" 973 974 def __init__(self, entries: Sequence[CompetitiveLandscapeEntry]): 975 if not entries: 976 raise ValueError("CompetitiveLandscape requires at least one entry") 977 self._entries = list(entries) 978 979 @classmethod 980 def from_file( 981 cls, path: str = COMPETITIVE_LANDSCAPE_DATASET 982 ) -> "CompetitiveLandscape": 983 with open(path, "r", encoding="utf-8") as handle: 984 payload = json.load(handle) 985 986 entries: List[CompetitiveLandscapeEntry] = [] 987 for record in payload: 988 entries.append( 989 CompetitiveLandscapeEntry( 990 competitor=record["competitor"], 991 channel=record["channel"], 992 summary=record["summary"], 993 strengths=list(record.get("strengths", [])), 994 weaknesses=list(record.get("weaknesses", [])), 995 recommended_play=record["recommended_play"], 996 share_of_voice=float(record.get("share_of_voice", 0.0)), 997 signal_intensity=float(record.get("signal_intensity", 0.0)), 998 keywords=[str(token).lower() for token in record.get("keywords", [])], 999 maturity=record.get("maturity", "growth"), 1000 source=record.get("source", "competitive_monitor"), 1001 notes=record.get("notes"), 1002 ) 1003 ) 1004 1005 return cls(entries) 1006 1007 @property 1008 def entries(self) -> List[CompetitiveLandscapeEntry]: 1009 return list(self._entries) 1010 1011 1012class LearningLoopProcess: 1013 """Transforms prompt intelligence into closed-loop learning recommendations.""" 1014 1015 def __init__(self, memory: PerformanceMemory | None = None): 1016 self._memory = memory or PerformanceMemory.from_file() 1017 1018 def recommend( 1019 self, 1020 prompt: str, 1021 diagnostics: "PromptDiagnostics", 1022 campaigns: Sequence["CampaignScore"], 1023 limit: int = 3, 1024 ) -> List[LearningLoopRecommendation]: 1025 tokens = MarketingIntelligenceEngine._keyword_set(prompt) 1026 diag_tokens: set[str] = set() 1027 for bucket in (diagnostics.opportunities, diagnostics.recommended_kpis): 1028 for item in bucket: 1029 diag_tokens.update(MarketingIntelligenceEngine._keyword_set(item)) 1030 1031 focus_tokens = set(MarketingIntelligenceEngine._keyword_set(diagnostics.focus)) 1032 urgency_bonus = 0.6 if diagnostics.urgency != "standard" else 0.0 1033 1034 channels = {score.campaign.channel.lower() for score in campaigns} 1035 audiences = {score.campaign.audience.lower() for score in campaigns} 1036 1037 scored: List[Tuple[float, PerformanceMemoryEntry]] = [] 1038 for entry in self._memory.entries: 1039 entry_tags = set(entry.tags) 1040 kpi_tokens: set[str] = set() 1041 for kpi in entry.kpi_focus: 1042 kpi_tokens.update(MarketingIntelligenceEngine._keyword_set(kpi)) 1043 combined_tags = entry_tags.union(kpi_tokens) 1044 1045 tag_hits = len(combined_tags.intersection(tokens)) 1046 diag_hits = len(combined_tags.intersection(diag_tokens)) 1047 focus_hits = len(combined_tags.intersection(focus_tokens)) 1048 1049 stage_bonus = 1.8 if entry.stage.lower() == diagnostics.stage.lower() else 0.0 1050 channel_bonus = 1.0 if entry.channel.lower() in channels else 0.0 1051 audience_bonus = 0.0 1052 for audience in audiences: 1053 if entry.audience_focus.lower() in audience or any(tag in audience for tag in combined_tags): 1054 audience_bonus = 0.9 1055 break 1056 1057 focus_bonus = 0.7 if focus_hits else 0.0 1058 opportunity_bonus = 0.5 * diag_hits 1059 keyword_bonus = 0.4 * tag_hits 1060 1061 score = ( 1062 entry.confidence 1063 + keyword_bonus 1064 + opportunity_bonus 1065 + focus_bonus 1066 + stage_bonus 1067 + channel_bonus 1068 + audience_bonus 1069 + urgency_bonus 1070 ) 1071 1072 scored.append((score, entry)) 1073 1074 ranked = sorted(scored, key=lambda item: item[0], reverse=True)[: max(1, limit)] 1075 1076 recommendations: List[LearningLoopRecommendation] = [] 1077 for _, entry in ranked: 1078 recommendations.append( 1079 LearningLoopRecommendation( 1080 play_name=entry.play_name, 1081 stage=entry.stage, 1082 audience_focus=entry.audience_focus, 1083 channel=entry.channel, 1084 learning=entry.summary, 1085 wins=entry.wins, 1086 losses=entry.losses, 1087 next_actions=entry.next_actions, 1088 kpi_focus=entry.kpi_focus, 1089 confidence=round(min(0.99, max(0.0, entry.confidence)), 2), 1090 source=entry.source, 1091 notes=entry.notes, 1092 ) 1093 ) 1094 1095 return recommendations 1096 1097 1098class CompetitiveIntelligenceProcess: 1099 """Turns competitive monitoring into counter-moves for the marketing ANI.""" 1100 1101 def __init__(self, landscape: CompetitiveLandscape | None = None): 1102 self._landscape = landscape or CompetitiveLandscape.from_file() 1103 1104 def signals( 1105 self, 1106 prompt: str, 1107 diagnostics: "PromptDiagnostics", 1108 campaigns: Sequence["CampaignScore"], 1109 limit: int = 3, 1110 ) -> List[CompetitiveSignal]: 1111 prompt_tokens = set(MarketingIntelligenceEngine._keyword_set(prompt)) 1112 focus_tokens = set(MarketingIntelligenceEngine._keyword_set(diagnostics.focus)) 1113 combined_tokens = prompt_tokens.union(focus_tokens) 1114 campaign_channels = {score.campaign.channel.lower() for score in campaigns} 1115 1116 stage_alignment = { 1117 "paid search": "conversion", 1118 "paid social": "consideration", 1119 "content syndication": "awareness", 1120 "outbound sdr": "conversion", 1121 "lifecycle automation": "retention", 1122 "partner ecosystem": "consideration", 1123 } 1124 1125 scored: List[Tuple[float, CompetitiveLandscapeEntry, float]] = [] 1126 for entry in self._landscape.entries: 1127 entry_tokens = set(entry.keywords) 1128 keyword_hits = len(entry_tokens.intersection(prompt_tokens)) 1129 focus_hits = len(entry_tokens.intersection(focus_tokens)) 1130 overlap_hits = len(entry_tokens.intersection(combined_tokens)) 1131 1132 share_component = entry.share_of_voice * 1.5 1133 intensity_component = entry.signal_intensity 1134 keyword_component = 0.25 * keyword_hits 1135 focus_component = 0.35 * focus_hits 1136 channel_component = 0.6 if entry.channel.lower() in campaign_channels else 0.0 1137 stage_tag = stage_alignment.get(entry.channel.lower()) 1138 stage_component = 0.4 if stage_tag == diagnostics.stage.lower() else 0.0 1139 maturity_component = 0.3 if entry.maturity.lower() in diagnostics.focus.lower() else 0.0 1140 1141 total_score = ( 1142 share_component 1143 + intensity_component 1144 + keyword_component 1145 + focus_component 1146 + channel_component 1147 + stage_component 1148 + maturity_component 1149 ) 1150 1151 focus_alignment = min(1.0, overlap_hits / max(1, len(entry_tokens))) 1152 1153 scored.append((total_score, entry, focus_alignment)) 1154 1155 ranked = sorted(scored, key=lambda item: item[0], reverse=True)[: max(1, limit)] 1156 1157 signals: List[CompetitiveSignal] = [] 1158 for total_score, entry, alignment in ranked: 1159 threat_index = entry.share_of_voice * 1.5 + entry.signal_intensity 1160 if threat_index >= 1.65: 1161 threat_level = "high" 1162 elif threat_index >= 1.15: 1163 threat_level = "medium" 1164 else: 1165 threat_level = "low" 1166 1167 signals.append( 1168 CompetitiveSignal( 1169 competitor=entry.competitor, 1170 channel=entry.channel, 1171 summary=entry.summary, 1172 strengths=entry.strengths, 1173 weaknesses=entry.weaknesses, 1174 recommended_play=entry.recommended_play, 1175 share_of_voice=round(entry.share_of_voice, 2), 1176 signal_intensity=round(entry.signal_intensity, 2), 1177 threat_level=threat_level, 1178 focus_alignment=round(alignment, 2), 1179 source=entry.source, 1180 notes=entry.notes, 1181 ) 1182 ) 1183 1184 return signals 1185 1186 1187class MarketingBenchmarks: 1188 """Loads benchmark statistics and produces campaign metrics.""" 1189 1190 def __init__(self, baseline: Dict[str, float], components: Dict[str, Dict[str, Dict[str, float]]]): 1191 self._baseline = baseline 1192 self._components = components 1193 1194 @classmethod 1195 def from_file(cls, path: str = MARKETING_BENCHMARKS) -> "MarketingBenchmarks": 1196 with open(path, "r", encoding="utf-8") as handle: 1197 payload = json.load(handle) 1198 1199 baseline = payload.get("baseline", {"roi": 2.0, "ctr": 0.02, "confidence": 0.6}) 1200 components: Dict[str, Dict[str, Dict[str, float]]] = {} 1201 for feature, entries in payload.items(): 1202 if feature == "baseline": 1203 continue 1204 components[feature] = {key.lower(): value for key, value in entries.items()} 1205 1206 return cls(baseline, components) 1207 1208 def _lookup(self, feature: str, value: str) -> Dict[str, float]: 1209 cohort = self._components.get(feature, {}) 1210 return cohort.get(value.lower(), self._baseline) 1211 1212 def metrics_for(self, campaign: MarketingCampaign) -> CampaignMetrics: 1213 weights = { 1214 "channels": 0.32, 1215 "audiences": 0.3, 1216 "offers": 0.22, 1217 "objectives": 0.16, 1218 } 1219 baseline_weight = max(0.0, 1.0 - sum(weights.values())) 1220 1221 base_roi = self._baseline.get("roi", 2.0) 1222 base_ctr = self._baseline.get("ctr", 0.02) 1223 base_conf = self._baseline.get("confidence", 0.6) 1224 1225 roi = base_roi * baseline_weight 1226 ctr = base_ctr * baseline_weight 1227 confidence = base_conf * baseline_weight 1228 drivers: Dict[str, float] = {} 1229 1230 features = { 1231 "channels": campaign.channel, 1232 "audiences": campaign.audience, 1233 "offers": campaign.offer, 1234 "objectives": campaign.objective, 1235 } 1236 1237 for feature, value in features.items(): 1238 stats = self._lookup(feature, value) 1239 weight = weights[feature] 1240 roi += stats.get("roi", base_roi) * weight 1241 ctr += stats.get("ctr", base_ctr) * weight 1242 confidence += stats.get("confidence", base_conf) * weight 1243 drivers[feature.rstrip("s")] = stats.get("roi", base_roi) 1244 1245 return CampaignMetrics( 1246 expected_roi=roi, 1247 expected_ctr=ctr, 1248 confidence=min(0.99, max(0.01, confidence)), 1249 drivers=drivers, 1250 ) 1251 1252 1253@dataclass(slots=True) 1254class CampaignScore: 1255 """Composite score for a marketing campaign.""" 1256 1257 campaign: MarketingCampaign 1258 score: float 1259 metrics: CampaignMetrics 1260 keyword_coverage: Dict[str, int] 1261 1262 1263@dataclass(slots=True) 1264class PromptDiagnostics: 1265 """Structured analysis of a marketing prompt.""" 1266 1267 stage: str 1268 urgency: str 1269 focus: str 1270 recommended_kpis: List[str] 1271 opportunities: List[str] 1272 risk_signals: List[str] 1273 keyword_summary: Dict[str, List[str]] 1274 confidence: float 1275 1276 1277@dataclass(slots=True) 1278class GrowthExperiment: 1279 """Experiment plan produced by the Experiment Growth Grid (EGG) process.""" 1280 1281 name: str 1282 hypothesis: str 1283 primary_metric: str 1284 secondary_metrics: List[str] 1285 playbook: List[str] 1286 confidence: float 1287 1288 1289@dataclass(slots=True) 1290class CJAffiliateOffer: 1291 """Offer metadata sourced from the Commission Junction catalog.""" 1292 1293 program_name: str 1294 advertiser_id: str 1295 category: str 1296 commission_rate: float 1297 epc: float 1298 seven_day_epc: float 1299 network_rank: int 1300 payout_terms: str 1301 landing_url: str 1302 description: str 1303 keywords: List[str] 1304 1305 1306@dataclass(slots=True) 1307class CJAffiliateRecommendation: 1308 """Recommendation returned by the CJ affiliate process.""" 1309 1310 program_name: str 1311 advertiser_id: str 1312 category: str 1313 commission_rate: float 1314 epc: float 1315 seven_day_epc: float 1316 projected_epc: float 1317 projected_monthly_revenue: float 1318 network_rank: int 1319 payout_terms: str 1320 landing_url: str 1321 rationale: str 1322 source: str 1323 notes: str | None = None 1324 1325 1326@dataclass(slots=True) 1327class InfluencerPartner: 1328 """Influencer metadata captured from curated partner intelligence.""" 1329 1330 name: str 1331 platform: str 1332 niche: str 1333 audience_size: int 1334 engagement_rate: float 1335 avg_cpm: float 1336 collab_formats: List[str] 1337 regions: List[str] 1338 conversion_focus: str 1339 keywords: List[str] 1340 case_study: str 1341 1342 1343@dataclass(slots=True) 1344class InfluencerRecommendation: 1345 """Recommendation returned by the influencer collaboration process.""" 1346 1347 name: str 1348 platform: str 1349 niche: str 1350 collab_format: str 1351 audience_size: int 1352 engagement_rate: float 1353 estimated_reach: int 1354 projected_cpm: float 1355 projected_leads: int 1356 projected_revenue: float 1357 projected_roi: float 1358 rationale: str 1359 source: str 1360 notes: str | None = None 1361 1362 1363class InfluencerNetworkCatalog: 1364 """Loads curated influencer partner records for offline orchestration.""" 1365 1366 def __init__(self, partners: Sequence[InfluencerPartner] | None = None) -> None: 1367 if partners is not None: 1368 self._partners = list(partners) 1369 else: 1370 with open(INFLUENCER_PARTNERS_DATASET, "r", encoding="utf-8") as handle: 1371 data = json.load(handle) 1372 self._partners = [InfluencerPartner(**entry) for entry in data] 1373 1374 def match_partners( 1375 self, tokens: Sequence[str], focus: str | None, limit: int 1376 ) -> List[Tuple[float, InfluencerPartner]]: 1377 token_set = {token.lower() for token in tokens} 1378 focus_tokens = {token.lower() for token in re.findall(r"[a-zA-Z0-9]+", focus or "")} 1379 1380 ranked: List[Tuple[float, InfluencerPartner]] = [] 1381 for partner in self._partners: 1382 haystack = ( 1383 f"{partner.name} {partner.platform} {partner.niche} {partner.conversion_focus} " 1384 f"{' '.join(partner.keywords)} {partner.case_study}" 1385 ).lower() 1386 score = 0.0 1387 keyword_hits = 0 1388 for token in token_set: 1389 if token in haystack: 1390 score += 1.8 1391 keyword_hits += 1 1392 if token in {keyword.lower() for keyword in partner.keywords}: 1393 score += 1.5 1394 keyword_hits += 1 1395 for token in focus_tokens: 1396 if token and token in haystack: 1397 score += 1.2 1398 if focus and focus.lower() in partner.conversion_focus.lower(): 1399 score += 1.0 1400 if focus and focus.lower() in partner.niche.lower(): 1401 score += 0.8 1402 score += min(partner.engagement_rate * 20, 2.4) 1403 score += min(partner.audience_size / 150000.0, 2.0) 1404 score += min(keyword_hits * 0.2, 1.5) 1405 if "ai" in token_set and "ai" in {k.lower() for k in partner.keywords}: 1406 score += 0.6 1407 ranked.append((score, partner)) 1408 1409 ranked.sort(key=lambda item: item[0], reverse=True) 1410 return ranked[: max(1, limit)] 1411 1412 1413@dataclass(slots=True) 1414class AmazonMarketplaceListing: 1415 """Amazon or AWS Marketplace listing metadata.""" 1416 1417 asin: str 1418 title: str 1419 category: str 1420 price_usd: float 1421 commission_rate: float 1422 rating: float 1423 reviews: int 1424 url: str 1425 keywords: List[str] 1426 estimated_monthly_sales: int 1427 buy_box_share: float 1428 prime_eligible: bool 1429 fulfillment_model: str 1430 1431 1432@dataclass(slots=True) 1433class AmazonProductRecommendation: 1434 """Recommendation payload produced by the Amazon marketplace process.""" 1435 1436 asin: str 1437 title: str 1438 category: str 1439 price_usd: float 1440 commission_rate: float 1441 rating: float 1442 reviews: int 1443 estimated_monthly_sales: int 1444 buy_box_share: float 1445 prime_eligible: bool 1446 fulfillment_model: str 1447 projected_monthly_revenue: float 1448 rationale: str 1449 url: str 1450 source: str 1451 notes: str | None = None 1452 1453 1454class AmazonMarketplaceCatalog: 1455 """Loads curated Amazon and AWS marketplace listings for offline use.""" 1456 1457 def __init__(self, listings: Sequence[AmazonMarketplaceListing] | None = None): 1458 if listings is not None: 1459 self._listings = list(listings) 1460 else: 1461 with open(AMAZON_MARKETPLACE_DATASET, "r", encoding="utf-8") as handle: 1462 data = json.load(handle) 1463 self._listings = [AmazonMarketplaceListing(**entry) for entry in data] 1464 1465 def match_listings( 1466 self, tokens: Sequence[str], limit: int, focus: str | None = None 1467 ) -> List[Tuple[float, AmazonMarketplaceListing]]: 1468 ranked: List[Tuple[float, AmazonMarketplaceListing]] = [] 1469 token_set = {token.lower() for token in tokens} 1470 for listing in self._listings: 1471 haystack = f"{listing.title} {listing.category} {' '.join(listing.keywords)}".lower() 1472 score = 0.0 1473 keyword_hits = 0 1474 for token in token_set: 1475 if token in haystack: 1476 score += 1.5 1477 keyword_hits += 1 1478 if token in {keyword.lower() for keyword in listing.keywords}: 1479 score += 1.8 1480 if focus: 1481 if focus.lower() in listing.category.lower(): 1482 score += 1.2 1483 if focus.lower() in haystack: 1484 score += 0.8 1485 score += min(listing.rating / 5.0, 1.0) 1486 score += min(listing.reviews / 500.0, 1.0) 1487 score += min(listing.estimated_monthly_sales / 400.0, 1.2) 1488 score += min(listing.buy_box_share, 1.0) 1489 if listing.prime_eligible: 1490 score += 0.35 1491 if listing.fulfillment_model.lower().startswith("fba"): 1492 score += 0.25 1493 score += keyword_hits * 0.1 1494 ranked.append((score, listing)) 1495 1496 ranked.sort(key=lambda item: item[0], reverse=True) 1497 return ranked[: max(1, limit)] 1498 1499 1500class AmazonMarketplaceProcess: