A continuation of Janady13/AI_Brain / services/seeder_api/intelligence.py. The complete source identity, every recorded occurrence, and linked equations are on its first page.
Source transcript · continued
Download complete source text ↓4,340 lines · 162.2 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
for env_var, key in cls.OPTIONAL_ENV_VARS.items():
value = os.getenv(env_var)
if value:
config[key] = value
return config
@property
def configured(self) -> bool:
return "api_key" in self._config
def recommendations(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int,
) -> List[InfluencerRecommendation]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
focus = None
if campaigns:
primary = campaigns[0].campaign
focus = f"{primary.channel} {primary.audience}"
matches = self._catalog.match_partners(tokens, focus, limit)
if not matches:
return []
avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.2
avg_ctr = mean([score.metrics.expected_ctr for score in campaigns]) if campaigns else 0.03
avg_confidence = (
mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.55
)
reach_multiplier = max(1.1, 1 + (avg_confidence - 0.5))
recs: List[InfluencerRecommendation] = []
for index, (match_score, partner) in enumerate(matches):
collab_formats = partner.collab_formats or ["sponsored integration"]
collab_format = collab_formats[index % len(collab_formats)]
estimated_reach = int(
round(partner.audience_size * partner.engagement_rate * reach_multiplier)
)
estimated_reach = max(estimated_reach, int(partner.audience_size * 0.02))
lead_rate = max(0.015, avg_ctr * 1.15 + partner.engagement_rate / 3)
projected_leads = max(12, int(round(estimated_reach * lead_rate)))
investment = partner.avg_cpm * estimated_reach / 1000.0
revenue_per_lead = max(95.0, avg_roi * 120.0)
projected_revenue = round(projected_leads * revenue_per_lead, 2)
projected_roi = (
round(projected_revenue / investment, 2) if investment else round(avg_roi * 3, 2)
)
keyword_matches = sorted(
{
token
for token in tokens
if token in {keyword.lower() for keyword in partner.keywords}
}
)
if keyword_matches:
rationale = (
"Maps "
+ "/".join(keyword_matches[:3])
+ f" intent into {partner.platform} {collab_format} with {partner.niche.lower()} authority."
)
else:
rationale = (
f"Pairs {partner.niche.lower()} reach with {partner.conversion_focus.lower()} focus "
"to amplify campaign demand."
)
if partner.case_study:
rationale += f" Proof: {partner.case_study}"
source = "influencer_api" if self.configured else "influencer_catalog"
notes = None
if not self.configured:
notes = (
"Simulated from influencer partner corpus; set INFLUENCER_NETWORK_API_KEY to "
"activate live network sourcing."
)
recs.append(
InfluencerRecommendation(
name=partner.name,
platform=partner.platform,
niche=partner.niche,
collab_format=collab_format,
audience_size=partner.audience_size,
engagement_rate=round(partner.engagement_rate, 4),
estimated_reach=estimated_reach,
projected_cpm=round(partner.avg_cpm, 2),
projected_leads=projected_leads,
projected_revenue=projected_revenue,
projected_roi=projected_roi,
rationale=rationale,
source=source,
notes=notes,
)
)
return recs
class AutomationSprintPlanner:
"""Builds a three-hour automation sprint tailored to the prompt context."""
def __init__(
self,
templates: Dict[str, Any] | None = None,
defaults: Dict[str, Any] | None = None,
):
data = templates or self._load_templates()
self._templates = data.get("stages", {})
self._overlays = data.get("focus_overlays", {})
base_defaults = data.get("defaults", {}).copy()
if defaults:
base_defaults.update(defaults)
self._defaults = base_defaults
@staticmethod
def _load_templates(path: str = AUTOMATION_SPRINTS_DATASET) -> Dict[str, Any]:
with open(path, "r", encoding="utf-8") as handle:
return json.load(handle)
def _stage_templates(self, stage: str) -> List[Dict[str, Any]]:
if stage in self._templates:
return list(self._templates[stage])
if "conversion" in self._templates:
return list(self._templates["conversion"])
for entries in self._templates.values():
if entries:
return list(entries)
return []
def _focus_overlays(self, focus: str) -> List[Dict[str, Any]]:
lowered = focus.lower()
for key, tasks in self._overlays.items():
if key in lowered:
return list(tasks)
return []
@staticmethod
def _safe_format(template: str, context: Dict[str, str]) -> str:
class _SafeDict(dict):
def __missing__(self, key: str) -> str: # pragma: no cover - defensive fallback
return "{" + key + "}"
return template.format_map(_SafeDict(context))
@staticmethod
def _campaign_context(
campaigns: Sequence[CampaignScore], index: int
) -> Dict[str, str]:
if campaigns:
primary = campaigns[0].campaign
reference = campaigns[index % len(campaigns)].campaign
secondary = campaigns[1 % len(campaigns)].campaign if len(campaigns) > 1 else primary
return {
"channel": reference.channel,
"offer": reference.offer,
"audience": reference.audience,
"objective": reference.objective,
"primary_channel": primary.channel,
"primary_offer": primary.offer,
"primary_audience": primary.audience,
"primary_objective": primary.objective,
"support_channel": secondary.channel,
"support_offer": secondary.offer,
}
return {
"channel": "omnichannel automation",
"offer": "flagship offer",
"audience": "core buyers",
"objective": "pipeline velocity",
"primary_channel": "omnichannel",
"primary_offer": "flagship offer",
"primary_audience": "core buyers",
"primary_objective": "pipeline velocity",
"support_channel": "email",
"support_offer": "activation play",
}
def plan(
self,
prompt: str,
diagnostics: PromptDiagnostics,
campaigns: Sequence[CampaignScore],
limit: int,
) -> AutomationSprint:
block = int(self._defaults.get("block_minutes", 30))
total_minutes = int(self._defaults.get("total_minutes", 180))
blocks = max(1, total_minutes // max(1, block))
templates = self._stage_templates(diagnostics.stage)
overlays = self._focus_overlays(diagnostics.focus)
if overlays:
templates.extend(overlays)
if not templates:
templates = [
{
"activity": "Audit acquisition funnel touchpoints",
"outcome": "Document automation opportunities",
"tooling": ["Notion"],
"kpis": ["pipeline velocity"],
"automation_level": "assist",
}
]
tasks: List[AutomationTask] = []
summary_drivers: List[str] = []
for idx in range(blocks):
template = templates[idx % len(templates)]
context = self._campaign_context(campaigns, idx)
context["stage"] = diagnostics.stage
context["focus"] = diagnostics.focus
activity = self._safe_format(template.get("activity", "Execute automation sprint"), context)
outcome = self._safe_format(template.get("outcome", "Drive measurable impact"), context)
tooling = list(template.get("tooling", []))
kpis = [kpi.lower() for kpi in template.get("kpis", [])]
automation_level = template.get("automation_level", "assist")
start_minute = idx * block
end_minute = start_minute + block
source = self._defaults.get("source", "automation_playbooks")
tasks.append(
AutomationTask(
start_minute=start_minute,
end_minute=end_minute,
activity=activity,
outcome=outcome,
tooling=tooling,
kpis=kpis,
automation_level=automation_level,
source=source,
)
)
channel = context.get("channel")
if channel and channel not in summary_drivers:
summary_drivers.append(channel)
channels_summary = ", ".join(summary_drivers[:3]) or "omnichannel"
summary = (
f"Three-hour automation sprint targeting {diagnostics.stage} intent for {diagnostics.focus} focus "
f"anchored in {channels_summary} plays."
)
notes = self._defaults.get("notes")
return AutomationSprint(
total_minutes=blocks * block,
stage=diagnostics.stage,
focus=diagnostics.focus,
summary=summary,
tasks=tasks,
notes=notes,
source=self._defaults.get("source", "automation_playbooks"),
)
class MarketingIntelligenceEngine:
"""Ranks marketing campaigns against free-form prompts."""
def __init__(
self,
campaigns: Sequence[MarketingCampaign],
benchmarks: MarketingBenchmarks | None = None,
google_process: GoogleAdsProcess | None = None,
cj_process: CJAffiliateProcess | None = None,
crypto_process: CryptoMarketProcess | None = None,
stripe_process: StripeProcess | None = None,
paypal_process: PayPalProcess | None = None,
amazon_process: AmazonMarketplaceProcess | None = None,
ebay_process: EbayMarketplaceProcess | None = None,
aws_process: AWSRevenueProcess | None = None,
budget_process: BudgetAllocatorProcess | None = None,
automation_process: AutomationSprintPlanner | None = None,
learning_process: LearningLoopProcess | None = None,
competitive_process: CompetitiveIntelligenceProcess | None = None,
influencer_process: InfluencerCollaborationProcess | None = None,
scenario_process: "ScenarioPlanningProcess" | None = None,
amazon_process: AmazonMarketplaceProcess | None = None,
aws_process: AWSRevenueProcess | None = None,
):
if not campaigns:
raise ValueError("MarketingIntelligenceEngine requires at least one campaign")
self._campaigns = list(campaigns)
self._benchmarks = benchmarks or MarketingBenchmarks.from_file()
self._google_process = google_process or GoogleAdsProcess()
self._cj_process = cj_process or CJAffiliateProcess()
self._crypto_process = crypto_process or CryptoMarketProcess()
self._stripe_process = stripe_process or StripeProcess()
self._paypal_process = paypal_process or PayPalProcess()
self._amazon_process = amazon_process or AmazonMarketplaceProcess()
self._ebay_process = ebay_process or EbayMarketplaceProcess()
self._aws_process = aws_process or AWSRevenueProcess()
self._budget_process = budget_process or BudgetAllocatorProcess()
self._automation_process = automation_process or AutomationSprintPlanner()
self._learning_process = learning_process or LearningLoopProcess()
self._competitive_process = competitive_process or CompetitiveIntelligenceProcess()
self._influencer_process = influencer_process or InfluencerCollaborationProcess()
self._scenario_process = scenario_process or ScenarioPlanningProcess()
self._amazon_process = amazon_process or AmazonMarketplaceProcess()
self._aws_process = aws_process or AWSRevenueProcess()
@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)
def cj_affiliate_enrichment(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[CJAffiliateRecommendation]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return []
return self._cj_process.offers(prompt, scored, limit)
def crypto_market_enrichment(
self,
prompt: str,
limit: int = 3,
) -> List[CryptoMarketSnapshot]:
return self._crypto_process.market_data(prompt, limit)
def stripe_checkout_enrichment(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[StripeMonetizationPlan]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return []
return self._stripe_process.plans(prompt, scored, limit)
def paypal_checkout_enrichment(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[PayPalMonetizationPlan]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return []
return self._paypal_process.plans(prompt, scored, limit)
def influencer_collaboration_enrichment(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[InfluencerRecommendation]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return []
return self._influencer_process.recommendations(prompt, scored, limit)
def amazon_marketplace_enrichment(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[AmazonProductRecommendation]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return []
return self._amazon_process.listings(prompt, scored, limit)
def ebay_marketplace_enrichment(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[EbayRevenuePlay]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return []
return self._ebay_process.listings(prompt, scored, limit)
def aws_cloud_enrichment(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[AWSRevenueRecommendation]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return []
return self._aws_process.plays(prompt, scored, limit)
def budget_optimizer(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> BudgetPlan:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return BudgetPlan(
total_budget=0.0,
allocations=[],
expected_revenue=0.0,
expected_roi=0.0,
payback_period_weeks=0.0,
summary="No campaigns available for budget allocation.",
)
return self._budget_process.plan(prompt, scored, limit)
def automation_sprint(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> AutomationSprint:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
diagnostics = self.diagnose(prompt)
return self._automation_process.plan(prompt, diagnostics, scored, limit)
def competitive_landscape(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[CompetitiveSignal]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
diagnostics = self.diagnose(prompt)
return self._competitive_process.signals(prompt, diagnostics, scored, limit)
def learning_loop(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[LearningLoopRecommendation]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
diagnostics = self.diagnose(prompt)
return self._learning_process.recommend(prompt, diagnostics, scored, limit)
def scenario_planning(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
*,
limit: int = 3,
budget: BudgetPlan | None = None,
google_ads: Sequence[GoogleCampaignPerformance] | None = None,
cj_affiliate: Sequence[CJAffiliateRecommendation] | None = None,
amazon: Sequence[AmazonProductRecommendation] | None = None,
ebay: Sequence[EbayRevenuePlay] | None = None,
aws: Sequence[AWSRevenueRecommendation] | None = None,
stripe: Sequence[StripeMonetizationPlan] | None = None,
paypal: Sequence[PayPalMonetizationPlan] | None = None,
crypto: Sequence[CryptoMarketSnapshot] | None = None,
influencers: Sequence[InfluencerRecommendation] | None = None,
) -> ScenarioAnalysis:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
diagnostics = self.diagnose(prompt)
return self._scenario_process.analyze(
prompt,
diagnostics,
scored,
budget=budget,
google_ads=google_ads or [],
cj_affiliate=cj_affiliate or [],
amazon=amazon or [],
ebay=ebay or [],
aws=aws or [],
stripe=stripe or [],
paypal=paypal or [],
crypto=crypto or [],
influencers=influencers or [],
limit=limit,
)
def excellence_summary(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
*,
limit: int,
rag: Sequence[MarketingCampaign],
egg: Sequence[GrowthExperiment],
google_ads: Sequence[GoogleCampaignPerformance],
cj_affiliate: Sequence[CJAffiliateRecommendation],
crypto_markets: Sequence[CryptoMarketSnapshot],
stripe: Sequence[StripeMonetizationPlan],
paypal: Sequence[PayPalMonetizationPlan],
amazon: Sequence[AmazonProductRecommendation],
ebay: Sequence[EbayRevenuePlay],
aws: Sequence[AWSRevenueRecommendation],
budget: BudgetPlan | None,
automation: AutomationSprint | None,
learning_loop: Sequence[LearningLoopRecommendation],
competitive: Sequence[CompetitiveSignal],
influencers: Sequence[InfluencerRecommendation],
) -> ExcellenceSummary:
limit = max(1, limit)
roi_avg = mean(entry.metrics.expected_roi for entry in campaigns) if campaigns else 0.0
ctr_avg = mean(entry.metrics.expected_ctr for entry in campaigns) if campaigns else 0.0
confidence_avg = (
mean(entry.metrics.confidence for entry in campaigns) if campaigns else 0.0
)
roi_norm = min(1.0, max(0.0, roi_avg / 5.0))
ctr_norm = min(1.0, max(0.0, ctr_avg / 0.05))
confidence_norm = min(1.0, max(0.0, confidence_avg))
coverage_map: Dict[str, float] = {
"rag": min(1.0, len(rag) / limit),
"egg": min(1.0, len(egg) / limit),
"google_ads": min(1.0, len(google_ads) / limit),
"cj_affiliate": min(1.0, len(cj_affiliate) / limit),
"crypto_markets": min(1.0, len(crypto_markets) / limit),
"stripe": min(1.0, len(stripe) / limit),
"paypal": min(1.0, len(paypal) / limit),
"influencer_collaborations": min(1.0, len(influencers) / limit),
"amazon_marketplace": min(1.0, len(amazon) / limit),
"ebay_marketplace": min(1.0, len(ebay) / limit),
"aws_cloud": min(1.0, len(aws) / limit),
"learning_loop": min(1.0, len(learning_loop) / limit),
"competitive_landscape": min(1.0, len(competitive) / limit),
}
coverage_map["budget_optimizer"] = 1.0 if budget and budget.allocations else 0.0
coverage_map["automation"] = 1.0 if automation and automation.tasks else 0.0
coverage_norm = mean(coverage_map.values()) if coverage_map else 0.0
excellence_index = round(
(roi_norm * 0.35 + ctr_norm * 0.2 + confidence_norm * 0.25 + coverage_norm * 0.2)
* 100,
2,
)
if excellence_index >= 85:
maturity = "world-class"
elif excellence_index >= 70:
maturity = "elite"
elif excellence_index >= 55:
maturity = "scaling"
else:
maturity = "emerging"
keywords = self.keywords(prompt)[:3]
if keywords:
headline_focus = ", ".join(keywords)
else:
headline_focus = "core growth themes"
headline = (
f"{maturity.title()} marketing ANI readiness anchored in {headline_focus}."
)
strengths: List[str] = []
if roi_norm >= 0.75:
strengths.append("Benchmark ROI trajectory rivals top performers.")
elif roi_norm >= 0.5:
strengths.append("ROI projections support confident scaling.")
if coverage_norm >= 0.7:
strengths.append("Process coverage spans key enrichment workflows.")
if len(competitive) >= 1:
strengths.append("Competitive counter-moves actively inform campaign plans.")
if len(influencers) >= 1:
strengths.append("Influencer collaborations unlock rapid audience amplification.")
watchouts: List[str] = []
if ctr_norm < 0.4:
watchouts.append(
"Click-through benchmarks trail aggressive targets; iterate creative experiments."
)
if confidence_norm < 0.6:
watchouts.append("Benchmark confidence is moderate; collect more lift validation.")
if coverage_norm < 0.5:
watchouts.append(
"Several enrichment processes returned limited data; review integration readiness."
)
if len(influencers) == 0:
watchouts.append("Source influencer partners to extend reach beyond owned channels.")
if not strengths:
strengths.append("Foundational datasets available for orchestration.")
if not watchouts:
watchouts.append("Continue monitoring live integrations to maintain excellence.")
notes = (
"Excellence index synthesizes ROI, CTR, confidence, and process coverage heuristics."
)
return ExcellenceSummary(
excellence_index=excellence_index,
maturity_level=maturity,
headline=headline,
strengths=strengths,
watchouts=watchouts,
process_coverage=coverage_map,
notes=notes,
)
def executive_playbook(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
*,
limit: int,
rag: Sequence[MarketingCampaign],
egg: Sequence[GrowthExperiment],
google_ads: Sequence[GoogleCampaignPerformance],
cj_affiliate: Sequence[CJAffiliateRecommendation],
amazon: Sequence[AmazonProductRecommendation],
ebay: Sequence[EbayRevenuePlay],
paypal: Sequence[PayPalMonetizationPlan],
budget: BudgetPlan | None,
automation: AutomationSprint | None,
learning_loop: Sequence[LearningLoopRecommendation],
competitive: Sequence[CompetitiveSignal],
influencers: Sequence[InfluencerRecommendation],
excellence: ExcellenceSummary | None = None,
) -> ExecutivePlaybook:
diagnostics = self.diagnose(prompt)
def _channel_key(text: str) -> str:
return re.sub(r"\s+", " ", text.lower().strip())
limit = max(1, limit)
top_scores = list(campaigns[: limit if campaigns else 1])
if not top_scores and campaigns:
top_scores = list(campaigns)
experiment_map: Dict[str, GrowthExperiment] = {}
for experiment in egg:
name_lower = experiment.name.lower()
channel_fragment: str | None = None
if " via " in name_lower:
channel_fragment = name_lower.split(" via ", 1)[1].strip()
if (not channel_fragment or not channel_fragment.strip()) and experiment.playbook:
first_step = experiment.playbook[0].lower()
match = re.search(r"activate (.+?) targeting", first_step)
if match:
channel_fragment = match.group(1).strip()
if channel_fragment:
experiment_map.setdefault(_channel_key(channel_fragment), experiment)
budget_map: Dict[str, BudgetAllocation] = {}
if budget:
for allocation in budget.allocations:
budget_map[_channel_key(allocation.channel)] = allocation
learning_map: Dict[str, List[LearningLoopRecommendation]] = {}
for recommendation in learning_loop:
channel_hint = recommendation.channel or recommendation.stage or recommendation.play_name
key = _channel_key(channel_hint)
learning_map.setdefault(key, []).append(recommendation)
competitive_map: Dict[str, List[CompetitiveSignal]] = {}
for signal in competitive:
competitive_map.setdefault(_channel_key(signal.channel), []).append(signal)
rag_channels = {idx: rag_campaign.channel for idx, rag_campaign in enumerate(rag)}
influencer_pool = list(influencers)
urgency_to_horizon = {"urgent": 72, "time-sensitive": 120, "standard": 168}
base_horizon = urgency_to_horizon.get(diagnostics.urgency, 168)
campaign_confidence = (
mean(score.metrics.confidence for score in top_scores) if top_scores else 0.5
)
roi_confidence = (
mean(score.metrics.expected_roi for score in top_scores) if top_scores else 0.5
)
budget_confidence = (
min(0.95, max(0.35, (budget.expected_roi / 5.0))) if budget else 0.55
)
learning_confidence = (
mean(rec.confidence for rec in learning_loop) if learning_loop else 0.55
)
automation_confidence = 0.75 if automation and automation.tasks else 0.5
overall_confidence = round(
min(
0.99,
max(
0.35,
mean(
[
diagnostics.confidence,
campaign_confidence,
min(0.95, max(0.3, roi_confidence / 5.0)),
budget_confidence,
learning_confidence,
automation_confidence,
]
),
),
),
2,
)
directives: List[StrategicDirective] = []
north_star_metric = (
diagnostics.recommended_kpis[0]
if diagnostics.recommended_kpis
else "revenue acceleration"
)
for index, score in enumerate(top_scores):
channel = score.campaign.channel
channel_key = _channel_key(channel)
allocation = budget_map.get(channel_key)
experiment = experiment_map.get(channel_key)
learnings = learning_map.get(channel_key, [])
competitors = competitive_map.get(channel_key, [])
google_signal = google_ads[index % len(google_ads)] if google_ads else None
affiliate_signal = cj_affiliate[index % len(cj_affiliate)] if cj_affiliate else None
amazon_signal = amazon[index % len(amazon)] if amazon else None
ebay_signal = ebay[index % len(ebay)] if ebay else None
paypal_signal = paypal[index % len(paypal)] if paypal else None
influencer_signal = (
influencer_pool[index % len(influencer_pool)] if influencer_pool else None
)
supporting_channels = [channel]
if google_signal:
supporting_channels.append("Google Ads")
if affiliate_signal:
supporting_channels.append("Affiliate")
if amazon_signal:
supporting_channels.append("Marketplace")
if ebay_signal:
supporting_channels.append("eBay")
if paypal_signal:
supporting_channels.append("PayPal")
if influencer_signal:
supporting_channels.append(f"Influencer - {influencer_signal.platform}")
if automation and automation.tasks:
supporting_channels.append("Automation Sprint")
supporting_channels = list(dict.fromkeys(supporting_channels))
action_steps: List[str] = [
(
f"Activate {channel} program targeting {score.campaign.audience} to "
f"{score.campaign.objective}."
)
]
if experiment:
action_steps.append(
f"Run Experiment '{experiment.name}' optimizing {experiment.primary_metric}."
)
if google_signal:
action_steps.append(
f"Model spend with Google Ads ROAS {google_signal.roas:.1f}x benchmark."
)
if affiliate_signal:
commission_pct = affiliate_signal.commission_rate * 100
action_steps.append(
f"Layer CJ affiliate offer '{affiliate_signal.program_name}' at {commission_pct:.1f}% commission."
)
if amazon_signal:
action_steps.append(
f"Promote marketplace listing '{amazon_signal.title}' to unlock partner revenue."
)
if ebay_signal:
action_steps.append(
(
f"Launch eBay offer '{ebay_signal.title}' leveraging {ebay_signal.affiliate_rate * 100:.1f}%"
" partner share and rapid sell-through."
)
)
if paypal_signal:
action_steps.append(
(
f"Enable PayPal {paypal_signal.checkout_flow} flow with {paypal_signal.payout_schedule} "
"payout cadence for dependable paychecks."
)
)
if influencer_signal:
action_steps.append(
(
f"Activate {influencer_signal.collab_format} with {influencer_signal.name} "
f"to reach {influencer_signal.estimated_reach:,.0f} engaged prospects."
)
)
if automation and automation.tasks:
first_task = automation.tasks[0]
kpi_focus = first_task.kpis[0] if first_task.kpis else "primary KPI"
action_steps.append(
f"Kick off automation sprint task '{first_task.activity}' for {kpi_focus}."
)
if learnings:
next_action = (
learnings[0].next_actions[0]
if learnings[0].next_actions
else "capture fresh experiment data"
)
action_steps.append(
f"Apply learning '{learnings[0].learning}' with next action '{next_action}'."
)
action_steps = list(dict.fromkeys(action_steps))
dependencies: List[str] = []
if allocation:
dependencies.append(
f"Allocate ${allocation.allocation_usd:,.0f} to {channel} with expected ${allocation.expected_revenue:,.0f} returns"
)
else:
dependencies.append("Confirm working budget envelope for channel execution")
if google_signal and google_signal.source != "simulated":
dependencies.append("Validate Google Ads account credentials and conversion tracking")
if affiliate_signal and affiliate_signal.source != "cj_catalog":
dependencies.append("Sync CJ affiliate program approvals")
if automation and automation.tasks:
dependencies.append("Reserve automation operators for 3-hour sprint")
if ebay_signal and ebay_signal.source != "ebay_catalog":
dependencies.append("Confirm eBay developer token and partner campaign tracking")
if paypal_signal and paypal_signal.source != "simulated":
dependencies.append("Verify PayPal payout configuration and webhook capture")
elif paypal_signal:
dependencies.append("Export PAYPAL credentials to activate live paychecks")
if influencer_signal:
est_investment = influencer_signal.projected_cpm * influencer_signal.estimated_reach / 1000.0
dependencies.append(
(
f"Fund influencer collaboration (~${est_investment:,.0f}) to secure "
f"{influencer_signal.projected_roi:.1f}x projected ROI"
)
)
risk_mitigations: List[str] = []
for risk in diagnostics.risk_signals[:2]:
risk_mitigations.append(f"Mitigate risk: {risk}")
for competitor in competitors[:1]:
risk_mitigations.append(
f"Counter {competitor.competitor} via {competitor.recommended_play}"
)
if learnings and learnings[0].losses:
risk_mitigations.append(
f"Avoid prior loss: {learnings[0].losses[0]}"
)
if influencer_signal and influencer_signal.projected_roi < 3.5:
risk_mitigations.append(
"Negotiate performance clauses with influencer partner to protect ROI"
)
if not risk_mitigations:
risk_mitigations.append("Maintain rapid feedback loops across processes")
directive_confidence = round(
min(
0.99,
max(
0.3,
mean(
[
score.metrics.confidence,
min(0.95, max(0.3, score.metrics.expected_roi / 5.0)),
learnings[0].confidence if learnings else diagnostics.confidence,
]
),
),
),
2,
)
time_horizon = base_horizon + index * 48
supporting_metrics_set: set[str] = {north_star_metric}
if experiment:
supporting_metrics_set.add(experiment.primary_metric)
supporting_metrics_set.update(experiment.secondary_metrics)
if influencer_signal:
supporting_metrics_set.add("influencer reach")
directive = StrategicDirective(
name=f"Directive {index + 1}: {score.campaign.channel.title()} Surge",
priority=index + 1,
objective=score.campaign.objective,
primary_channel=channel,
supporting_channels=supporting_channels,
investment_usd=round(allocation.allocation_usd, 2) if allocation else None,
expected_revenue_usd=round(allocation.expected_revenue, 2) if allocation else None,
confidence=directive_confidence,
time_horizon_hours=time_horizon,
north_star_metric=north_star_metric,
supporting_metrics=sorted(supporting_metrics_set),
action_steps=action_steps,
dependencies=dependencies,
risk_mitigations=risk_mitigations,
inspiration=" | ".join(
filter(
None,
[
experiment.name if experiment else None,
learnings[0].play_name if learnings else None,
competitors[0].competitor if competitors else None,
rag_channels.get(index),
],
)
)
or "Campaign intelligence fusion",
)
directives.append(directive)
summary = (
f"Synthesize {len(directives)} directives to accelerate {diagnostics.stage} outcomes "
f"for {diagnostics.focus}."
)
notes_candidates = []
if excellence:
notes_candidates.append(excellence.headline)
if budget:
notes_candidates.append(budget.summary)
if automation:
notes_candidates.append(automation.summary)
if influencer_pool:
notes_candidates.append(
(
f"Influencer plan anchored by {influencer_pool[0].name} "
f"({influencer_pool[0].projected_roi:.1f}x ROI outlook)"
)
)
notes = " | ".join(notes_candidates) if notes_candidates else None
return ExecutivePlaybook(
headline=
f"{diagnostics.stage.title()} momentum for {diagnostics.focus} with autonomous orchestration",
focus=diagnostics.focus,
urgency=diagnostics.urgency,
north_star_metric=north_star_metric,
total_budget_usd=round(budget.total_budget, 2) if budget else None,
expected_revenue_usd=round(budget.expected_revenue, 2) if budget else None,
confidence=overall_confidence,
directives=directives,
summary=summary,
notes=notes,
)
@lru_cache(maxsize=1)
def default_marketing_engine() -> MarketingIntelligenceEngine:
return MarketingIntelligenceEngine.from_file()