"""Marketing intelligence engine that ranks and scores enumerated campaigns.""" from __future__ import annotations import os import re import json from dataclasses import dataclass from functools import lru_cache from typing import Dict, List, Sequence, Tuple try: # pragma: no cover - optional dependency that may not be installed in tests from google.ads.googleads.client import GoogleAdsClient # type: ignore except ImportError: # pragma: no cover - gracefully handled when SDK missing GoogleAdsClient = None # type: ignore MARKETING_ENUMERATION = os.path.join( os.path.dirname(__file__), "..", "..", "daemon", "enumerations", "marketing_500.txt" ) MARKETING_BENCHMARKS = os.path.join( os.path.dirname(__file__), "..", "..", "daemon", "datasets", "marketing_benchmarks.json" ) def _load_lines(path: str) -> List[str]: with open(path, "r", encoding="utf-8") as handle: return [line.strip() for line in handle if line.strip()] @dataclass(slots=True) class MarketingCampaign: """Structured representation for one marketing campaign idea.""" raw: str channel: str audience: str offer: str objective: str @classmethod def parse(cls, raw: str) -> "MarketingCampaign": text = raw.strip().rstrip(".") if not text.lower().startswith("campaign:"): raise ValueError(f"Unsupported campaign format: {raw}") body = text[len("Campaign:") :].strip() try: channel, remainder = body.split("targeting", 1) audience, remainder = remainder.split("leveraging", 1) offer, objective = remainder.split(" to ", 1) except ValueError as exc: # pragma: no cover - enumeration should be well formed raise ValueError(f"Malformed campaign sentence: {raw}") from exc return cls( raw=raw, channel=channel.strip(), audience=audience.strip(), offer=offer.strip(), objective=objective.strip().rstrip("."), ) @dataclass(slots=True) class CampaignMetrics: """Statistical expectations for a marketing campaign.""" expected_roi: float expected_ctr: float confidence: float drivers: Dict[str, float] class MarketingBenchmarks: """Loads benchmark statistics and produces campaign metrics.""" def __init__(self, baseline: Dict[str, float], components: Dict[str, Dict[str, Dict[str, float]]]): self._baseline = baseline self._components = components @classmethod def from_file(cls, path: str = MARKETING_BENCHMARKS) -> "MarketingBenchmarks": with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) baseline = payload.get("baseline", {"roi": 2.0, "ctr": 0.02, "confidence": 0.6}) components: Dict[str, Dict[str, Dict[str, float]]] = {} for feature, entries in payload.items(): if feature == "baseline": continue components[feature] = {key.lower(): value for key, value in entries.items()} return cls(baseline, components) def _lookup(self, feature: str, value: str) -> Dict[str, float]: cohort = self._components.get(feature, {}) return cohort.get(value.lower(), self._baseline) def metrics_for(self, campaign: MarketingCampaign) -> CampaignMetrics: weights = { "channels": 0.32, "audiences": 0.3, "offers": 0.22, "objectives": 0.16, } baseline_weight = max(0.0, 1.0 - sum(weights.values())) base_roi = self._baseline.get("roi", 2.0) base_ctr = self._baseline.get("ctr", 0.02) base_conf = self._baseline.get("confidence", 0.6) roi = base_roi * baseline_weight ctr = base_ctr * baseline_weight confidence = base_conf * baseline_weight drivers: Dict[str, float] = {} features = { "channels": campaign.channel, "audiences": campaign.audience, "offers": campaign.offer, "objectives": campaign.objective, } for feature, value in features.items(): stats = self._lookup(feature, value) weight = weights[feature] roi += stats.get("roi", base_roi) * weight ctr += stats.get("ctr", base_ctr) * weight confidence += stats.get("confidence", base_conf) * weight drivers[feature.rstrip("s")] = stats.get("roi", base_roi) return CampaignMetrics( expected_roi=roi, expected_ctr=ctr, confidence=min(0.99, max(0.01, confidence)), drivers=drivers, ) @dataclass(slots=True) class CampaignScore: """Composite score for a marketing campaign.""" campaign: MarketingCampaign score: float metrics: CampaignMetrics keyword_coverage: Dict[str, int] @dataclass(slots=True) class PromptDiagnostics: """Structured analysis of a marketing prompt.""" stage: str urgency: str focus: str recommended_kpis: List[str] opportunities: List[str] risk_signals: List[str] keyword_summary: Dict[str, List[str]] confidence: float @dataclass(slots=True) class GrowthExperiment: """Experiment plan produced by the Experiment Growth Grid (EGG) process.""" name: str hypothesis: str primary_metric: str secondary_metrics: List[str] playbook: List[str] confidence: float @dataclass(slots=True) class GoogleCampaignPerformance: """Performance snapshot surfaced from the Google Ads API.""" campaign_name: str clicks: int impressions: int conversions: float cost_micros: int ctr: float cpc: float roas: float source: str notes: str | None = None class GoogleAdsProcess: """Bridges the Google Ads API to enrich ANI recommendations.""" REQUIRED_ENV_VARS = { "GOOGLE_ADS_DEVELOPER_TOKEN": "developer_token", "GOOGLE_ADS_CLIENT_ID": "client_id", "GOOGLE_ADS_CLIENT_SECRET": "[credential removed]", "GOOGLE_ADS_REFRESH_TOKEN": "refresh_token", "GOOGLE_ADS_CUSTOMER_ID": "customer_id", } OPTIONAL_ENV_VARS = { "GOOGLE_ADS_LOGIN_CUSTOMER_ID": "login_customer_id", } def __init__(self, config: Dict[str, str] | None = None): self._config = config or self._config_from_env() self._client = self._build_client(self._config) @classmethod def _config_from_env(cls) -> Dict[str, str]: config: Dict[str, str] = {} for env_var, key in cls.REQUIRED_ENV_VARS.items(): value = os.getenv(env_var) if value: config[key] = value for env_var, key in cls.OPTIONAL_ENV_VARS.items(): value = os.getenv(env_var) if value: config[key] = value return config @staticmethod def _build_client(config: Dict[str, str]) -> GoogleAdsClient | None: if not config or "customer_id" not in config: return None if GoogleAdsClient is None: return None try: # pragma: no cover - relies on network credentials to execute client = GoogleAdsClient.load_from_dict( { "developer_token": config.get("developer_token"), "client_id": config.get("client_id"), "client_secret": config.get("client_secret"), "refresh_token": config.get("refresh_token"), "login_customer_id": config.get("login_customer_id"), "use_proto_plus": True, } ) return client except Exception: # pragma: no cover - fallback handled downstream return None @property def configured(self) -> bool: return bool(self._client) def performance( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int, ) -> List[GoogleCampaignPerformance]: limit = max(1, limit) if self._client is None: return self._simulate(prompt, campaigns, limit) try: # pragma: no cover - executed when Google Ads credentials available ga_service = self._client.get_service("GoogleAdsService") query = self._build_query(limit) response = ga_service.search( customer_id=self._config.get("customer_id"), query=query, ) insights: List[GoogleCampaignPerformance] = [] for row in response: metrics = row.metrics campaign = row.campaign clicks = int(getattr(metrics, "clicks", 0)) impressions = int(getattr(metrics, "impressions", 0)) conversions = float(getattr(metrics, "conversions", 0.0)) cost_micros = int(getattr(metrics, "cost_micros", 0)) ctr = float(getattr(metrics, "ctr", 0.0)) cpc = float(getattr(metrics, "average_cpc", 0.0)) roas = float(getattr(metrics, "conversions_value_per_cost", 0.0)) insights.append( GoogleCampaignPerformance( campaign_name=getattr(campaign, "name", "Campaign"), clicks=clicks, impressions=impressions, conversions=conversions, cost_micros=cost_micros, ctr=ctr, cpc=cpc, roas=roas, source="google_ads_api", notes="Live Google Ads data", ) ) if insights: return insights except Exception: # pragma: no cover - gracefully fallback on API errors pass return self._simulate(prompt, campaigns, limit) @staticmethod def _build_query(limit: int) -> str: return ( "SELECT campaign.name, metrics.clicks, metrics.impressions, metrics.conversions, " "metrics.cost_micros, metrics.ctr, metrics.average_cpc, metrics.conversions_value_per_cost " "FROM campaign WHERE segments.date DURING LAST_30_DAYS " "ORDER BY metrics.conversions DESC " f"LIMIT {limit}" ) def _simulate( self, prompt: str, campaigns: Sequence[CampaignScore], limit: int ) -> List[GoogleCampaignPerformance]: simulated: List[GoogleCampaignPerformance] = [] tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower()) emphasis = 1 + len(set(tokens)) * 0.02 for index, scored in enumerate(campaigns[:limit]): base_clicks = max(50, int(scored.score * 10)) clicks = int(base_clicks * emphasis) impressions = max(clicks * 8, 500) conversions = round(max(1.0, scored.metrics.expected_roi * 1.5), 2) cost_micros = int(max(1_000_000, impressions * 120)) ctr = round(min(0.25, clicks / max(1, impressions)), 4) cpc = round(cost_micros / max(1, clicks) / 1_000_000, 2) roas = round(scored.metrics.expected_roi / max(0.1, cpc), 2) simulated.append( GoogleCampaignPerformance( campaign_name=f"{scored.campaign.channel.title()} - Simulated", clicks=clicks, impressions=impressions, conversions=conversions, cost_micros=cost_micros, ctr=ctr, cpc=cpc, roas=roas, source="simulated", notes=( "Simulated from marketing benchmarks; set GOOGLE_ADS_* env vars " "for live Google Ads enrichment." ), ) ) if not simulated: simulated.append( GoogleCampaignPerformance( campaign_name="Simulated Campaign", clicks=120, impressions=2400, conversions=3.0, cost_micros=2500000, ctr=0.05, cpc=0.5, roas=4.2, source="simulated", notes="Fallback simulation due to missing campaigns", ) ) return simulated class MarketingIntelligenceEngine: """Ranks marketing campaigns against free-form prompts.""" def __init__( self, campaigns: Sequence[MarketingCampaign], benchmarks: MarketingBenchmarks | None = None, google_process: GoogleAdsProcess | None = None, ): if not campaigns: raise ValueError("MarketingIntelligenceEngine requires at least one campaign") self._campaigns = list(campaigns) self._benchmarks = benchmarks or MarketingBenchmarks.from_file() self._google_process = google_process or GoogleAdsProcess() @classmethod def from_file( cls, path: str = MARKETING_ENUMERATION, benchmark_path: str = MARKETING_BENCHMARKS, ) -> "MarketingIntelligenceEngine": lines = _load_lines(path) campaigns = [MarketingCampaign.parse(line) for line in lines] benchmarks = MarketingBenchmarks.from_file(benchmark_path) return cls(campaigns, benchmarks) @staticmethod def _keyword_set(prompt: str) -> List[str]: tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower()) filtered = [token for token in tokens if len(token) > 2] return filtered or ["campaign"] def recommend(self, prompt: str, limit: int = 3) -> List[MarketingCampaign]: tokens = self._keyword_set(prompt) def score(campaign: MarketingCampaign) -> tuple[int, int, str]: haystacks = { "channel": campaign.channel.lower(), "audience": campaign.audience.lower(), "offer": campaign.offer.lower(), "objective": campaign.objective.lower(), } weighted = 0 matched_features: set[str] = set() for token in tokens: for feature, text in haystacks.items(): if token in text: matched_features.add(feature) weight = 4 if feature == "audience" else 3 if feature == "offer" else 2 weighted += weight # fallback to slight preference for broader coverage even if nothing matched coverage_bonus = len(matched_features) return (-(weighted + coverage_bonus), len(campaign.raw), campaign.raw) ranked = sorted(self._campaigns, key=score) return ranked[: max(1, limit)] def recommend_statistical(self, prompt: str, limit: int = 3) -> List[CampaignScore]: tokens = self._keyword_set(prompt) scored: List[CampaignScore] = [] for campaign in self._campaigns: metrics = self._benchmarks.metrics_for(campaign) keyword_hits: Dict[str, int] = {"channel": 0, "audience": 0, "offer": 0, "objective": 0} haystacks = { "channel": campaign.channel.lower(), "audience": campaign.audience.lower(), "offer": campaign.offer.lower(), "objective": campaign.objective.lower(), } match_bonus = 0.0 for token in tokens: for feature, text in haystacks.items(): if token in text: keyword_hits[feature] += 1 match_bonus += 0.35 roi_component = metrics.expected_roi * 0.6 ctr_component = metrics.expected_ctr * 80.0 * 0.25 confidence_component = metrics.confidence * 10.0 * 0.1 keyword_component = match_bonus composite = roi_component + ctr_component + confidence_component + keyword_component scored.append( CampaignScore( campaign=campaign, score=composite, metrics=metrics, keyword_coverage=keyword_hits, ) ) ranked = sorted(scored, key=lambda entry: entry.score, reverse=True) return ranked[: max(1, limit)] def summarize(self, prompt: str, campaigns: Sequence[MarketingCampaign]) -> str: tokens = self._keyword_set(prompt) highlights = ", ".join(sorted(set(tokens))) count = len(campaigns) return f"Surface {count} campaign plays tuned for {highlights}." @staticmethod def _infer_stage(tokens: Sequence[str]) -> Tuple[str, List[str]]: mapping = { "awareness": {"awareness", "reach", "impressions", "brand", "launch", "attention"}, "consideration": {"consideration", "education", "webinar", "nurture", "pipeline", "engagement"}, "conversion": { "conversion", "roi", "ctr", "leads", "demos", "trial", "customers", "acquisition", }, "retention": {"retention", "renewal", "expansion", "upsell", "loyalty", "churn"}, } hits: Dict[str, List[str]] = {key: [] for key in mapping} for token in tokens: for stage, keywords in mapping.items(): if token in keywords: hits[stage].append(token) stage = max(mapping, key=lambda key: len(hits[key])) if all(len(values) == 0 for values in hits.values()): stage = "conversion" return stage, hits[stage] @staticmethod def _infer_urgency(prompt: str) -> str: lowered = prompt.lower() if any(keyword in lowered for keyword in ["urgent", "asap", "immediately", "right away", "now"]): return "urgent" if any(keyword in lowered for keyword in ["this quarter", "q4", "deadline", "accelerate"]): return "time-sensitive" return "standard" @staticmethod def _infer_focus(tokens: Sequence[str]) -> Tuple[str, List[str]]: focus_buckets = { "enterprise b2b": {"enterprise", "cxo", "c-suite", "fortune", "b2b", "saas"}, "startup founders": {"founder", "startup", "scaleup", "seed", "series"}, "consumer": {"consumer", "b2c", "retail", "shopper", "app"}, "developer": {"developer", "devops", "engineer", "technical", "api"}, } hits: Dict[str, List[str]] = {bucket: [] for bucket in focus_buckets} for token in tokens: for focus, keywords in focus_buckets.items(): if token in keywords: hits[focus].append(token) focus = max(focus_buckets, key=lambda key: len(hits[key])) if len(hits[focus]) == 0: return "broad b2b", [] return focus, hits[focus] @staticmethod def _recommend_kpis(stage: str) -> List[str]: mapping = { "awareness": ["impressions", "share of voice", "reach growth"], "consideration": ["content engagement", "marketing qualified leads", "pipeline velocity"], "conversion": ["sales qualified leads", "demo-to-close rate", "customer acquisition cost"], "retention": ["net revenue retention", "renewal rate", "expansion ARR"], } return mapping.get(stage, ["pipeline velocity", "conversion rate"]) @staticmethod def _recommend_opportunities(stage: str, focus: str) -> List[str]: opportunities = { "awareness": [ "Anchor thought-leadership content to prime the market", "Deploy programmatic amplification across hero stories", ], "consideration": [ "Introduce nurture sequences aligned to buyer pains", "Layer intent scoring to prioritize engaged accounts", ], "conversion": [ "Bundle conversion offers with social proof assets", "Tighten sales enablement loops with marketing triggers", ], "retention": [ "Launch customer marketing to reinforce product value", "Pair success cadences with expansion playbooks", ], } base = opportunities.get(stage, []) if focus.startswith("enterprise"): base = base + ["Align executive-level narratives with ROI benchmarks"] elif "startup" in focus: base = base + ["Stand up scrappy experiments to validate fast wins"] elif "consumer" in focus: base = base + ["Personalize lifecycle messaging using behavioral cohorts"] return base @staticmethod def _detect_risks(prompt: str) -> List[str]: lowered = prompt.lower() risks: List[str] = [] if "flat" in lowered or "stalled" in lowered or "plateau" in lowered: risks.append("Performance plateau signaled") if "limited budget" in lowered or "tight budget" in lowered: risks.append("Budget constraints highlighted") if "new market" in lowered or "unknown" in lowered: risks.append("Market unfamiliarity risk") if "churn" in lowered or "retention" in lowered: risks.append("Retention fragility noted") return risks def diagnose(self, prompt: str) -> PromptDiagnostics: tokens = self._keyword_set(prompt) stage, stage_hits = self._infer_stage(tokens) urgency = self._infer_urgency(prompt) focus, focus_hits = self._infer_focus(tokens) kpis = self._recommend_kpis(stage) opportunities = self._recommend_opportunities(stage, focus) risks = self._detect_risks(prompt) keyword_summary = { "intent": stage_hits, "focus": focus_hits, "tokens": tokens, } confidence_seed = len(stage_hits) + len(focus_hits) confidence = max(0.35, min(0.95, 0.35 + 0.1 * confidence_seed)) return PromptDiagnostics( stage=stage, urgency=urgency, focus=focus, recommended_kpis=kpis, opportunities=opportunities, risk_signals=risks, keyword_summary=keyword_summary, confidence=round(confidence, 2), ) def portfolio(self, prompt: str, campaigns: Sequence[CampaignScore]) -> CampaignMetrics: if not campaigns: raise ValueError("portfolio requires at least one campaign score") roi = sum(entry.metrics.expected_roi for entry in campaigns) / len(campaigns) ctr = sum(entry.metrics.expected_ctr for entry in campaigns) / len(campaigns) confidence = sum(entry.metrics.confidence for entry in campaigns) / len(campaigns) drivers: Dict[str, float] = {} for entry in campaigns: for feature, value in entry.metrics.drivers.items(): drivers[feature] = max(drivers.get(feature, 0.0), value) return CampaignMetrics( expected_roi=roi, expected_ctr=ctr, confidence=confidence, drivers=drivers, ) def keywords(self, prompt: str) -> List[str]: return self._keyword_set(prompt) def expansion_grid(self, prompt: str, campaigns: Sequence[CampaignScore]) -> List[GrowthExperiment]: """Design a set of Experiment Growth Grid (EGG) plays from campaign rankings.""" if not campaigns: raise ValueError("expansion_grid requires at least one campaign score") diagnostics = self.diagnose(prompt) stage_recipes: Dict[str, List[Dict[str, str]]] = { "awareness": [ { "name": "Narrative Amplification Sprint", "hypothesis": ( "If we spotlight {offer} through {channel}, " "we will warm {audience} toward {objective}." ), "play": "Launch hero content and amplify across paid/owned reach streams.", "metric": "impressions", }, { "name": "Thought Leadership Cascade", "hypothesis": ( "Positioning {audience} challenges alongside {offer} " "via {channel} will expand top-of-funnel attention." ), "play": "Sequence authoritative narratives across social and partner syndication.", "metric": "share of voice", }, ], "consideration": [ { "name": "Intent Magnet Program", "hypothesis": ( "If we orchestrate {offer} journeys through {channel}, " "we can mature {audience} into evaluators for {objective}." ), "play": "Build nurture modules triggered by behavioral intent signals.", "metric": "marketing qualified leads", }, { "name": "Education Sprint", "hypothesis": ( "Providing {audience} with {offer} across {channel} " "will compress evaluation cycles." ), "play": "Layer webinars, playbooks, and retargeting to reinforce learnings.", "metric": "content engagement", }, ], "conversion": [ { "name": "Pipeline Acceleration Lab", "hypothesis": ( "If {audience} experience {offer} via {channel}, " "we can unlock {objective} faster." ), "play": "Run multi-touch conversion blitz with calibrated social proof.", "metric": "sales qualified leads", }, { "name": "Revenue Experiment Track", "hypothesis": ( "Blending {offer} within {channel} for {audience} will uplift deal velocity." ), "play": "Pair high-intent triggers with offer experimentation and CRO support.", "metric": "conversion rate", }, ], "retention": [ { "name": "Value Expansion Studio", "hypothesis": ( "If {audience} engage with {offer} programs via {channel}, " "we will fortify {objective}." ), "play": "Launch customer storytelling and advocacy loops to reinforce value.", "metric": "renewal rate", }, { "name": "Lifecycle Momentum Cell", "hypothesis": ( "Activating {offer} through {channel} for {audience} " "will catalyze expansion outcomes." ), "play": "Coordinate success-led campaigns with product usage nudges.", "metric": "net revenue retention", }, ], } focus_overlays: Dict[str, Dict[str, str]] = { "enterprise b2b": { "play": "Map executive-specific proof points and ABM orchestration layers.", "metric": "executive meeting volume", }, "startup founders": { "play": "Prototype scrappy experiments with rapid iteration cadences.", "metric": "activation-to-demo ratio", }, "consumer": { "play": "Employ personalization matrices and lifecycle messaging bursts.", "metric": "repeat purchase rate", }, "developer": { "play": "Enable technical champions with sandboxes, tutorials, and community proof.", "metric": "trial-to-adoption rate", }, "default": { "play": "Instrument experiment tracking with weekly conversion retros.", "metric": "conversion rate", }, } base_recipes = stage_recipes.get(diagnostics.stage, stage_recipes["conversion"]) overlay = focus_overlays.get(diagnostics.focus, focus_overlays["default"]) recommended = diagnostics.recommended_kpis or ["conversion rate"] experiments: List[GrowthExperiment] = [] for index, scored in enumerate(campaigns[: max(1, len(campaigns))]): recipe = base_recipes[index % len(base_recipes)] name = f"{recipe['name']} via {scored.campaign.channel.title()}" hypothesis = recipe["hypothesis"].format( offer=scored.campaign.offer, channel=scored.campaign.channel, audience=scored.campaign.audience, objective=scored.campaign.objective, ) playbook = [ f"Activate {scored.campaign.channel} targeting {scored.campaign.audience}.", recipe["play"], overlay["play"], f"Align offer '{scored.campaign.offer}' to accelerate {scored.campaign.objective} outcomes.", ] primary_metric = recipe.get("metric", recommended[0]) secondary_candidates = set(recommended[1:]) secondary_candidates.add(overlay["metric"]) secondary_metrics = sorted(secondary_candidates) confidence = (diagnostics.confidence + scored.metrics.confidence) / 2 experiments.append( GrowthExperiment( name=name, hypothesis=hypothesis, primary_metric=primary_metric, secondary_metrics=secondary_metrics, playbook=playbook, confidence=round(max(0.2, min(0.99, confidence)), 2), ) ) return experiments def google_ads_enrichment( self, prompt: str, campaigns: Sequence[CampaignScore] | None = None, limit: int = 3, ) -> List[GoogleCampaignPerformance]: scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit) if not scored: return [] return self._google_process.performance(prompt, scored, limit) @lru_cache(maxsize=1) def default_marketing_engine() -> MarketingIntelligenceEngine: return MarketingIntelligenceEngine.from_file()