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