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 ↓5,617 lines · 212.6 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
"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 statistical_reasoning_summary(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
*,
budget: BudgetPlan | None,
stock_markets: Sequence[StockMarketForecast] | None,
crypto_markets: Sequence[CryptoMarketSnapshot] | None,
limit: int,
) -> StatisticalReasoningSummary:
return self._statistical_process.analyze(
prompt,
campaigns,
budget=budget,
stock_markets=stock_markets,
crypto_markets=crypto_markets,
limit=limit,
)
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 stock_market_enrichment(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[StockMarketForecast]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return []
return self._stock_process.forecasts(prompt, scored, 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 cognitive_architecture_enrichment(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[CognitiveCapabilityPlan]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return []
return self._cognitive_process.plans(prompt, scored, limit)
def agentic_system_enrichment(
self,
prompt: str,
campaigns: Sequence[CampaignScore] | None = None,
limit: int = 3,
) -> List[AgenticSystemPlan]:
scored = list(campaigns) if campaigns is not None else self.recommend_statistical(prompt, limit)
if not scored:
return []
return self._agentic_process.plans(prompt, 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,
stocks: Sequence[StockMarketForecast] | None = None,
statistics: StatisticalReasoningSummary | None = None,
cognitive: Sequence[CognitiveCapabilityPlan] | None = None,
agentic: Sequence[AgenticSystemPlan] | 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 [],
stocks=stocks or [],
statistics=statistics,
cognitive=cognitive or [],
agentic=agentic 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],
stocks: Sequence[StockMarketForecast],
statistics: StatisticalReasoningSummary | None = None,
cognitive: Sequence[CognitiveCapabilityPlan] | None = None,
agentic: Sequence[AgenticSystemPlan] | None = None,
) -> 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),
"stock_markets": min(1.0, len(stocks) / 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_map["statistical_reasoning"] = 1.0 if statistics else 0.0
coverage_map["cognitive_architecture"] = (
min(1.0, len(cognitive or []) / limit) if cognitive is not None else 0.0
)
coverage_map["agentic_systems"] = (
min(1.0, len(agentic or []) / limit) if agentic is not None 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.")
if stocks:
top_stock = stocks[0]
if top_stock.forecast_return_90d > 0.1:
strengths.append(
f"{top_stock.name} outlook adds capital market tailwind to marketing scale."
)
if statistics and statistics.success_probability >= 0.75:
strengths.append(
f"Statistical reasoning signals {statistics.success_probability:.0%} success probability."
)
if cognitive:
top_capability = cognitive[0]
if top_capability.maturity in {"advanced", "autonomous"}:
strengths.append(
f"Cognitive architecture running {top_capability.maturity} on {top_capability.capability}."
)
if agentic:
system = agentic[0]
strengths.append(
f"Agentic system {system.system_name} orchestrates {system.maturity} tool calling stack."
)
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 stocks and mean(forecast.volatility for forecast in stocks) > 0.25:
watchouts.append("Equity volatility elevated; stage reinvestment with safeguards.")
if statistics and statistics.success_probability < 0.6:
watchouts.append(
"Statistical reasoning indicates moderate success probability; reinforce testing cadence."
)
if not cognitive:
watchouts.append(
"Activate cognitive architecture plans so reasoning and learning stay orchestrated."
)
if not agentic:
watchouts.append("Wire agentic systems to automate tool calling and database access.")
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."
)
if stocks:
top_stock = stocks[0]
notes += (
f" | Stock outlook {top_stock.symbol} {top_stock.trend} with "
f"{top_stock.forecast_return_30d*100:.1f}% 30d move."
)
if statistics:
notes += f" | {statistics.narrative}"
if cognitive:
notes += (
f" | Cognitive architecture anchored by {cognitive[0].capability} ({cognitive[0].maturity})."
)
if agentic:
notes += (
f" | Agentic system {agentic[0].system_name} ({agentic[0].maturity}) governing tool orchestration."
)
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],
stocks: Sequence[StockMarketForecast],
excellence: ExcellenceSummary | None = None,
statistics: StatisticalReasoningSummary | None = None,
cognitive: Sequence[CognitiveCapabilityPlan] | None = None,
agentic: Sequence[AgenticSystemPlan] | 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
statistics_confidence = statistics.confidence if statistics else 0.6
agentic_confidence = 0.78 if agentic 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,
(mean(stock.confidence for stock in stocks) if stocks else 0.6),
statistics_confidence,
agentic_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
)
stock_signal = stocks[index % len(stocks)] if stocks else None
cognitive_plan = cognitive[index % len(cognitive)] if cognitive else None
agentic_plan = agentic[index % len(agentic)] if agentic 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 stock_signal:
supporting_channels.append(f"Markets - {stock_signal.symbol}")
if automation and automation.tasks:
supporting_channels.append("Automation Sprint")
if statistics:
supporting_channels.append("Statistical Reasoning")
if cognitive:
supporting_channels.append(f"Cognitive - {cognitive[0].capability}")
if agentic:
supporting_channels.append(f"Agentic - {agentic[0].system_name}")
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 stock_signal:
action_steps.append(
(
f"Align budget pacing with {stock_signal.name} {stock_signal.trend} trend "
f"({stock_signal.forecast_return_30d*100:.1f}% 30d outlook)."
)
)
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 statistics:
top_driver = max(
statistics.driver_confidence.items(),
key=lambda item: item[1],
)[0].split(":", 1)[1] if statistics.driver_confidence else "core drivers"
action_steps.append(
f"Instrument statistical driver '{top_driver}' to validate {statistics.expected_roi:.2f}x ROI forecast."
)
if cognitive_plan:
action_steps.append(
(
f"Operationalize {cognitive_plan.capability} via {cognitive_plan.frameworks[0]} "
"to keep ANI reasoning adaptive."
)
)
if agentic_plan:
primary_tool = agentic_plan.toolchain[0] if agentic_plan.toolchain else "agent orchestrator"
action_steps.append(
(
f"Deploy agentic system {agentic_plan.system_name} using {primary_tool} "
"to unify tool calling and database flows."
)
)
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"
)
)
if stock_signal:
dependencies.append(
(
f"Track {stock_signal.symbol} volatility ({stock_signal.volatility:.2f}) "
"to schedule reinvestment windows."
)
)
if statistics and statistics.success_probability < 0.7:
dependencies.append(
"Schedule mid-cycle statistical refresh to bolster success probability"
)
if cognitive_plan:
dependencies.append(
(
f"Deploy {cognitive_plan.capability} ops with {cognitive_plan.tooling[0]} instrumentation"
)
)
if agentic_plan:
dependencies.append(
(
f"Enable {agentic_plan.system_name} access to {agentic_plan.data_interfaces[0] if agentic_plan.data_interfaces else 'core data systems'}"
)
)
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 stock_signal and stock_signal.volatility > 0.28:
risk_mitigations.append(
f"Hedge equity exposure while {stock_signal.symbol} volatility runs {stock_signal.volatility:.2f}"
)
if statistics and statistics.volatility_adjustment > 0.25:
risk_mitigations.append(
"Stage spend releases to account for statistical volatility adjustments"
)
if cognitive_plan and cognitive_plan.maturity == "foundational":
risk_mitigations.append(
f"Invest in advancing {cognitive_plan.capability} maturity for sustained autonomy"
)
if agentic_plan and agentic_plan.maturity in {"foundational", "advanced"}:
risk_mitigations.append(
f"Upskill agentic system {agentic_plan.system_name} workflows for higher autonomy"
)
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")
if stock_signal:
supporting_metrics_set.add("market return correlation")
if statistics:
supporting_metrics_set.add("statistical success probability")
if cognitive_plan:
supporting_metrics_set.add("cognitive readiness index")
if agentic_plan:
supporting_metrics_set.add(agentic_plan.metrics[0] if agentic_plan.metrics else "agentic readiness")
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)"
)
)
if stocks:
top_stock = stocks[0]
notes_candidates.append(
(
f"Market outlook: {top_stock.name} {top_stock.trend} with "
f"{top_stock.forecast_return_90d*100:.0f}% 90d upside"
)
)
if statistics:
notes_candidates.append(statistics.narrative)
if cognitive:
notes_candidates.append(
f"Cognitive architecture anchored by {cognitive[0].capability} ({cognitive[0].maturity})."
)
if agentic:
notes_candidates.append(
f"Agentic system {agentic[0].system_name} ({agentic[0].maturity}) orchestrates tool stack."
)
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()