Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/AI_Brain / services/seeder_api/intelligence.py |
| Date | Not specified in source |
| Repository | Janady13/AI_Brain |
| Revision | 49db80a641b10ab62d16cc7643bd0b367f93951c |
| Branch | codex/build-online-github-ani |
| Path or archive member | services/seeder_api/intelligence.py |
| Original SHA-256 | 3cb2fed20f0ff185b19f1c150c2b4a334491f5ef807c165dca83bf81dbd89d4c |
| Git blob SHA | fca36e8ec6208d02c78f97edebae7b019ce9c740 |
| Reading-copy SHA-256 | 0676c5a0fbf5ee584efaefc9a10331ca2a85c079bb35afcc60988f9a748dc7f2 |
Recovered source transcript
Download complete source text ↓789 lines · 30.3 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
"""Marketing intelligence engine that ranks and scores enumerated campaigns."""
from __future__ import annotations
import 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()