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 ↓9,479 lines · 364.7 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
@staticmethod
def _stabilizers(
diagnostics: PromptDiagnostics,
budget: BudgetPlan | None,
statistics: StatisticalReasoningSummary | None,
cognitive: Sequence[CognitiveCapabilityPlan] | None,
agentic: Sequence[AgenticSystemPlan] | None,
knowledge: Sequence[KnowledgeWorkflowPlan] | None,
language: Sequence[LanguageCapabilityPlan] | None,
code_automation: Sequence[CodeAutomationPlan] | None,
payments: Sequence[PaymentIntegrationPlan] | None,
investing: Sequence[InvestingPlan] | None,
agentic_loops: Sequence[AgenticLoopPlan] | None,
quantum: Sequence[QuantumStressTestPlan] | None,
physics_economics: Sequence[PhysicsEconomicsPlan] | None,
meta_learning: Sequence[MetaLearningPlan] | None,
marketing_knowledge: Sequence[MarketingKnowledgePlan] | None,
) -> List[str]:
stabilizers: List[str] = []
if diagnostics.recommended_kpis:
stabilizers.append(f"Instrument KPI guardrails around {diagnostics.recommended_kpis[0]}")
if budget and budget.allocations:
top_channel = budget.allocations[0].channel
stabilizers.append(f"Phase spend across {top_channel} with weekly optimization cycles")
if statistics:
stabilizers.append(
f"Recompute statistical priors weekly to maintain {statistics.success_probability:.0%} success guardrails"
)
if cognitive:
stabilizers.append(
f"Audit cognitive capability '{cognitive[0].capability}' to preserve prompt and reasoning quality"
)
if language:
stabilizers.append(
f"Maintain language workflow '{language[0].capability}' so prompts and transcripts stay machine readable"
)
if code_automation:
stabilizers.append(
f"Run post-deploy audits on {code_automation[0].capability} to keep automation safe"
)
if payments:
stabilizers.append(
f"Document payment runbooks for {payments[0].provider} to safeguard go-live troubleshooting"
)
if investing:
stabilizers.append(
f"Publish investing compliance guardrails for {investing[0].provider} campaigns"
)
if agentic:
stabilizers.append(
f"Schedule agentic system '{agentic[0].system_name}' health checks to keep tool calling compliant"
)
if knowledge:
stabilizers.append("Refresh RAG/KAG indices weekly to maintain knowledge coverage >90%.")
if marketing_knowledge:
stabilizers.append(
f"Maintain {marketing_knowledge[0].name} sync cadence ({marketing_knowledge[0].update_frequency})."
)
if agentic_loops:
stabilizers.append("Run loop retrospectives every 12 hours to capture learning deltas.")
if quantum:
stabilizers.append(
f"Review quantum stress telemetry from {quantum[0].simulator} before scaling spend."
)
if physics_economics:
stabilizers.append(
f"Audit physics/economics plan {physics_economics[0].capability} cadence to preserve algorithmic fidelity."
)
if meta_learning:
stabilizers.append(
f"Review meta-learning loop '{meta_learning[0].capability}' checkpoints for safe adaptation."
)
return stabilizers
def analyze(
self,
prompt: str,
diagnostics: PromptDiagnostics,
campaigns: Sequence[CampaignScore],
*,
budget: BudgetPlan | None,
google_ads: Sequence[GoogleCampaignPerformance] | None,
cj_affiliate: Sequence[CJAffiliateRecommendation] | None,
amazon: Sequence[AmazonProductRecommendation] | None,
ebay: Sequence[EbayRevenuePlay] | None,
aws: Sequence[AWSRevenueRecommendation] | None,
stripe: Sequence[StripeMonetizationPlan] | None,
paypal: Sequence[PayPalMonetizationPlan] | None,
payments: Sequence[PaymentIntegrationPlan] | None,
investing: Sequence[InvestingPlan] | None,
crypto: Sequence[CryptoMarketSnapshot] | None,
influencers: Sequence[InfluencerRecommendation] | None,
stocks: Sequence[StockMarketForecast] | None,
statistics: StatisticalReasoningSummary | None,
cognitive: Sequence[CognitiveCapabilityPlan] | None,
agentic: Sequence[AgenticSystemPlan] | None,
knowledge: Sequence[KnowledgeWorkflowPlan] | None,
marketing_knowledge: Sequence[MarketingKnowledgePlan] | None,
code_automation: Sequence[CodeAutomationPlan] | None,
language: Sequence[LanguageCapabilityPlan] | None,
meta_learning: Sequence[MetaLearningPlan] | None,
moe: Sequence[MixtureOfExpertsPlan] | None,
agentic_loops: Sequence[AgenticLoopPlan] | None,
quantum: Sequence[QuantumStressTestPlan] | None,
physics_economics: Sequence[PhysicsEconomicsPlan] | None,
limit: int = 3,
) -> ScenarioAnalysis:
if not campaigns:
raise ValueError("Scenario planning requires campaign scores")
top = list(campaigns[: max(1, limit)])
agentic = list(agentic or [])
investing = list(investing or [])
meta_learning = list(meta_learning or [])
quantum = list(quantum or [])
physics_economics = list(physics_economics or [])
marketing_knowledge = list(marketing_knowledge or [])
avg_roi = mean(score.metrics.expected_roi for score in top)
avg_confidence = mean(score.metrics.confidence for score in top)
avg_ctr = mean(score.metrics.expected_ctr for score in top)
base_budget = (
budget.total_budget if budget and budget.total_budget > 0 else self._estimate_budget(diagnostics, top)
)
base_revenue = (
budget.expected_revenue
if budget and budget.expected_revenue > 0
else round(base_budget * max(1.1, avg_roi), 2)
)
base_roi = round(base_revenue / base_budget, 2) if base_budget else round(avg_roi, 2)
if statistics:
inferred_budget = (
statistics.expected_revenue / max(statistics.expected_roi, 0.1)
if statistics.expected_roi > 0
else base_budget
)
base_budget = max(base_budget, inferred_budget)
base_revenue = max(base_revenue, statistics.expected_revenue)
base_roi = max(base_roi, statistics.expected_roi)
volatility = self._volatility_floor
if diagnostics.urgency == "urgent":
volatility += 0.1
elif diagnostics.urgency == "time-sensitive":
volatility += 0.05
if diagnostics.risk_signals:
volatility += 0.08
if crypto and any(snapshot.change_24h and abs(snapshot.change_24h) > 5 for snapshot in crypto):
volatility += 0.05
if stocks and any(forecast.volatility > 0.3 for forecast in stocks):
volatility += 0.04
elif stocks and mean(forecast.volatility for forecast in stocks) < 0.12:
volatility -= 0.02
if quantum:
quantum_stability = mean(test.stability_index for plan in quantum for test in plan.stress_tests)
if quantum_stability < 0.45:
volatility += 0.05
elif quantum_stability > 0.6:
volatility -= 0.02
if statistics:
volatility = max(volatility, statistics.volatility_adjustment)
volatility = min(0.5, max(self._volatility_floor, volatility))
optimistic_multiplier = 1 + volatility
conservative_multiplier = max(0.45, 1 - volatility)
optimistic_revenue = round(base_revenue * optimistic_multiplier, 2)
conservative_revenue = round(base_revenue * conservative_multiplier, 2)
optimistic_roi = round(base_roi * (1 + volatility / 1.5), 2)
conservative_roi = round(base_roi * (1 - volatility / 1.2), 2)
optimistic_conf = min(0.99, avg_confidence + (0.12 - volatility / 2))
base_conf = min(0.95, max(0.25, avg_confidence))
conservative_conf = max(0.2, avg_confidence - volatility / 2)
optimistic_prob = max(0.18, 0.32 - (volatility - self._volatility_floor) * 0.6)
conservative_prob = max(0.22, 0.28 + (volatility - self._volatility_floor) * 0.5)
base_prob = max(0.3, 1.0 - optimistic_prob - conservative_prob)
if statistics:
blended_base = statistics.success_probability * 0.6 + base_prob * 0.4
base_prob = min(0.85, max(0.25, blended_base))
residual = max(0.15, 1.0 - base_prob)
share_denominator = max(optimistic_prob + conservative_prob, 0.01)
optimistic_share = optimistic_prob / share_denominator
optimistic_prob = min(0.6, residual * optimistic_share)
conservative_prob = max(0.1, residual - optimistic_prob)
total = optimistic_prob + base_prob + conservative_prob
optimistic_prob = round(max(0.05, optimistic_prob / total), 2)
base_prob = round(max(0.1, base_prob / total), 2)
conservative_prob = round(max(0.05, conservative_prob / total), 2)
accelerators = self._signal_accelerators(
google_ads=google_ads,
cj_affiliate=cj_affiliate,
amazon=amazon,
ebay=ebay,
aws=aws,
stripe=stripe,
paypal=paypal,
payments=payments,
investing=investing,
influencers=influencers,
stocks=stocks,
statistics=statistics,
cognitive=cognitive,
agentic=agentic,
knowledge=knowledge,
marketing_knowledge=marketing_knowledge,
code_automation=code_automation,
language=language,
meta_learning=meta_learning,
moe=moe,
agentic_loops=agentic_loops,
quantum=quantum,
physics_economics=physics_economics,
)
stabilizers = self._stabilizers(
diagnostics,
budget,
statistics,
cognitive,
agentic,
knowledge,
language,
code_automation,
payments,
investing,
agentic_loops,
quantum,
physics_economics,
meta_learning,
marketing_knowledge,
)
if avg_ctr > 0.6:
accelerators.append("Benchmark CTR momentum indicates creative resonance")
if len(accelerators) < 2:
accelerators.append("Cross-process enrichment unlocks multi-channel leverage")
if not stabilizers:
stabilizers.append("Institute weekly analytics reviews to course-correct")
risk_watchouts = diagnostics.risk_signals[:2]
if not risk_watchouts:
risk_watchouts = ["Monitor acquisition efficiency as spend scales"]
projections = [
ScenarioProjection(
scenario="optimistic",
expected_revenue=optimistic_revenue,
expected_roi=optimistic_roi,
probability=optimistic_prob,
confidence=round(optimistic_conf, 2),
accelerators=accelerators,
stabilizers=stabilizers,
watchouts=["Ensure fulfillment capacity scales with demand"],
),
ScenarioProjection(
scenario="base",
expected_revenue=round(base_revenue, 2),
expected_roi=round(base_roi, 2),
probability=base_prob,
confidence=round(base_conf, 2),
accelerators=accelerators,
stabilizers=stabilizers,
watchouts=risk_watchouts,
),
ScenarioProjection(
scenario="conservative",
expected_revenue=conservative_revenue,
expected_roi=round(conservative_roi, 2),
probability=round(conservative_prob, 2),
confidence=round(conservative_conf, 2),
accelerators=accelerators,
stabilizers=stabilizers,
watchouts=["Reserve contingency budget to reallocate into proven channels"],
),
]
primary_channels = {score.campaign.channel for score in top[:2]}
summary = (
f"Model optimistic, base, and conservative revenue trajectories for {diagnostics.stage} "
f"outcomes targeting {diagnostics.focus}."
)
headline = (
f"Scenario coverage for {', '.join(sorted(primary_channels))} anchored in {diagnostics.focus}."
)
notes = (
f"Volatility factor {volatility:.2f} derives from urgency={diagnostics.urgency} and risk signals"
)
if statistics:
notes += (
f" | Statistical reasoning forecasts {statistics.expected_roi:.2f}x ROI with "
f"{statistics.success_probability:.0%} success probability"
)
if agentic:
notes += (
f" | Agentic system {agentic[0].system_name} ({agentic[0].maturity}) orchestrates tool calling"
)
if code_automation:
notes += (
f" | Code automation {code_automation[0].capability} ({code_automation[0].maturity}) holds deployment guardrails"
)
return ScenarioAnalysis(
headline=headline,
summary=summary,
projections=projections,
confidence=round(base_conf, 2),
accelerators=accelerators,
stabilizers=stabilizers,
notes=notes,
)
class PerformanceMemory:
"""Loads and scores historical marketing learnings."""
def __init__(self, entries: Sequence[PerformanceMemoryEntry]):
if not entries:
raise ValueError("PerformanceMemory requires at least one entry")
self._entries = list(entries)
@classmethod
def from_file(cls, path: str = PERFORMANCE_MEMORY_DATASET) -> "PerformanceMemory":
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
entries = []
for record in payload:
entries.append(
PerformanceMemoryEntry(
play_name=record["play_name"],
stage=record["stage"],
audience_focus=record["audience_focus"],
channel=record["channel"],
tags=[tag.lower() for tag in record.get("tags", [])],
summary=record["summary"],
wins=record.get("wins", []),
losses=record.get("losses", []),
next_actions=record.get("next_actions", []),
kpi_focus=[str(kpi) for kpi in record.get("kpi_focus", [])],
confidence=float(record.get("confidence", 0.5)),
source=record.get("source", "performance_memory"),
notes=record.get("notes"),
)
)
return cls(entries)
@property
def entries(self) -> List[PerformanceMemoryEntry]:
return list(self._entries)
@dataclass(slots=True)
class CompetitiveLandscapeEntry:
"""Competitive signal captured from market monitoring."""
competitor: str
channel: str
summary: str
strengths: List[str]
weaknesses: List[str]
recommended_play: str
share_of_voice: float
signal_intensity: float
keywords: List[str]
maturity: str
source: str
notes: str | None = None
@dataclass(slots=True)
class CompetitiveSignal:
"""Recommendation derived from the competitive landscape dataset."""
competitor: str
channel: str
summary: str
strengths: List[str]
weaknesses: List[str]
recommended_play: str
share_of_voice: float
signal_intensity: float
threat_level: str
focus_alignment: float
source: str
notes: str | None = None
class CompetitiveLandscape:
"""Loads competitive marketing signals from curated monitoring."""
def __init__(self, entries: Sequence[CompetitiveLandscapeEntry]):
if not entries:
raise ValueError("CompetitiveLandscape requires at least one entry")
self._entries = list(entries)
@classmethod
def from_file(
cls, path: str = COMPETITIVE_LANDSCAPE_DATASET
) -> "CompetitiveLandscape":
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
entries: List[CompetitiveLandscapeEntry] = []
for record in payload:
entries.append(
CompetitiveLandscapeEntry(
competitor=record["competitor"],
channel=record["channel"],
summary=record["summary"],
strengths=list(record.get("strengths", [])),
weaknesses=list(record.get("weaknesses", [])),
recommended_play=record["recommended_play"],
share_of_voice=float(record.get("share_of_voice", 0.0)),
signal_intensity=float(record.get("signal_intensity", 0.0)),
keywords=[str(token).lower() for token in record.get("keywords", [])],
maturity=record.get("maturity", "growth"),
source=record.get("source", "competitive_monitor"),
notes=record.get("notes"),
)
)
return cls(entries)
@property
def entries(self) -> List[CompetitiveLandscapeEntry]:
return list(self._entries)
class LearningLoopProcess:
"""Transforms prompt intelligence into closed-loop learning recommendations."""
def __init__(self, memory: PerformanceMemory | None = None):
self._memory = memory or PerformanceMemory.from_file()
def recommend(
self,
prompt: str,
diagnostics: "PromptDiagnostics",
campaigns: Sequence["CampaignScore"],
limit: int = 3,
) -> List[LearningLoopRecommendation]:
tokens = MarketingIntelligenceEngine._keyword_set(prompt)
diag_tokens: set[str] = set()
for bucket in (diagnostics.opportunities, diagnostics.recommended_kpis):
for item in bucket:
diag_tokens.update(MarketingIntelligenceEngine._keyword_set(item))
focus_tokens = set(MarketingIntelligenceEngine._keyword_set(diagnostics.focus))
urgency_bonus = 0.6 if diagnostics.urgency != "standard" else 0.0
channels = {score.campaign.channel.lower() for score in campaigns}
audiences = {score.campaign.audience.lower() for score in campaigns}
scored: List[Tuple[float, PerformanceMemoryEntry]] = []
for entry in self._memory.entries:
entry_tags = set(entry.tags)
kpi_tokens: set[str] = set()
for kpi in entry.kpi_focus:
kpi_tokens.update(MarketingIntelligenceEngine._keyword_set(kpi))
combined_tags = entry_tags.union(kpi_tokens)
tag_hits = len(combined_tags.intersection(tokens))
diag_hits = len(combined_tags.intersection(diag_tokens))
focus_hits = len(combined_tags.intersection(focus_tokens))
stage_bonus = 1.8 if entry.stage.lower() == diagnostics.stage.lower() else 0.0
channel_bonus = 1.0 if entry.channel.lower() in channels else 0.0
audience_bonus = 0.0
for audience in audiences:
if entry.audience_focus.lower() in audience or any(tag in audience for tag in combined_tags):
audience_bonus = 0.9
break
focus_bonus = 0.7 if focus_hits else 0.0
opportunity_bonus = 0.5 * diag_hits
keyword_bonus = 0.4 * tag_hits
score = (
entry.confidence
+ keyword_bonus
+ opportunity_bonus
+ focus_bonus
+ stage_bonus
+ channel_bonus
+ audience_bonus
+ urgency_bonus
)
scored.append((score, entry))
ranked = sorted(scored, key=lambda item: item[0], reverse=True)[: max(1, limit)]
recommendations: List[LearningLoopRecommendation] = []
for _, entry in ranked:
recommendations.append(
LearningLoopRecommendation(
play_name=entry.play_name,
stage=entry.stage,
audience_focus=entry.audience_focus,
channel=entry.channel,
learning=entry.summary,
wins=entry.wins,
losses=entry.losses,
next_actions=entry.next_actions,
kpi_focus=entry.kpi_focus,
confidence=round(min(0.99, max(0.0, entry.confidence)), 2),
source=entry.source,
notes=entry.notes,
)
)
return recommendations
class CompetitiveIntelligenceProcess:
"""Turns competitive monitoring into counter-moves for the marketing ANI."""
def __init__(self, landscape: CompetitiveLandscape | None = None):
self._landscape = landscape or CompetitiveLandscape.from_file()
def signals(
self,
prompt: str,
diagnostics: "PromptDiagnostics",
campaigns: Sequence["CampaignScore"],
limit: int = 3,
) -> List[CompetitiveSignal]:
prompt_tokens = set(MarketingIntelligenceEngine._keyword_set(prompt))
focus_tokens = set(MarketingIntelligenceEngine._keyword_set(diagnostics.focus))
combined_tokens = prompt_tokens.union(focus_tokens)
campaign_channels = {score.campaign.channel.lower() for score in campaigns}
stage_alignment = {
"paid search": "conversion",
"paid social": "consideration",
"content syndication": "awareness",
"outbound sdr": "conversion",
"lifecycle automation": "retention",
"partner ecosystem": "consideration",
}
scored: List[Tuple[float, CompetitiveLandscapeEntry, float]] = []
for entry in self._landscape.entries:
entry_tokens = set(entry.keywords)
keyword_hits = len(entry_tokens.intersection(prompt_tokens))
focus_hits = len(entry_tokens.intersection(focus_tokens))
overlap_hits = len(entry_tokens.intersection(combined_tokens))
share_component = entry.share_of_voice * 1.5
intensity_component = entry.signal_intensity
keyword_component = 0.25 * keyword_hits
focus_component = 0.35 * focus_hits
channel_component = 0.6 if entry.channel.lower() in campaign_channels else 0.0
stage_tag = stage_alignment.get(entry.channel.lower())
stage_component = 0.4 if stage_tag == diagnostics.stage.lower() else 0.0
maturity_component = 0.3 if entry.maturity.lower() in diagnostics.focus.lower() else 0.0
total_score = (
share_component
+ intensity_component
+ keyword_component
+ focus_component
+ channel_component
+ stage_component
+ maturity_component
)
focus_alignment = min(1.0, overlap_hits / max(1, len(entry_tokens)))
scored.append((total_score, entry, focus_alignment))
ranked = sorted(scored, key=lambda item: item[0], reverse=True)[: max(1, limit)]
signals: List[CompetitiveSignal] = []
for total_score, entry, alignment in ranked:
threat_index = entry.share_of_voice * 1.5 + entry.signal_intensity
if threat_index >= 1.65:
threat_level = "high"
elif threat_index >= 1.15:
threat_level = "medium"
else:
threat_level = "low"
signals.append(
CompetitiveSignal(
competitor=entry.competitor,
channel=entry.channel,
summary=entry.summary,
strengths=entry.strengths,
weaknesses=entry.weaknesses,
recommended_play=entry.recommended_play,
share_of_voice=round(entry.share_of_voice, 2),
signal_intensity=round(entry.signal_intensity, 2),
threat_level=threat_level,
focus_alignment=round(alignment, 2),
source=entry.source,
notes=entry.notes,
)
)
return signals
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 CJAffiliateOffer:
"""Offer metadata sourced from the Commission Junction catalog."""
program_name: str
advertiser_id: str
category: str
commission_rate: float
epc: float
seven_day_epc: float
network_rank: int
payout_terms: str
landing_url: str
description: str
keywords: List[str]
@dataclass(slots=True)
class CJAffiliateRecommendation:
"""Recommendation returned by the CJ affiliate process."""
program_name: str
advertiser_id: str
category: str
commission_rate: float
epc: float
seven_day_epc: float
projected_epc: float
projected_monthly_revenue: float
network_rank: int
payout_terms: str
landing_url: str
rationale: str
source: str
notes: str | None = None
@dataclass(slots=True)
class InfluencerPartner:
"""Influencer metadata captured from curated partner intelligence."""
name: str
platform: str
niche: str
audience_size: int
engagement_rate: float
avg_cpm: float
collab_formats: List[str]
regions: List[str]
conversion_focus: str
keywords: List[str]
case_study: str
@dataclass(slots=True)
class InfluencerRecommendation:
"""Recommendation returned by the influencer collaboration process."""
name: str
platform: str
niche: str
collab_format: str
audience_size: int
engagement_rate: float
estimated_reach: int
projected_cpm: float
projected_leads: int
projected_revenue: float
projected_roi: float
rationale: str
source: str
notes: str | None = None
class InfluencerNetworkCatalog:
"""Loads curated influencer partner records for offline orchestration."""
def __init__(self, partners: Sequence[InfluencerPartner] | None = None) -> None:
if partners is not None:
self._partners = list(partners)
else:
with open(INFLUENCER_PARTNERS_DATASET, "r", encoding="utf-8") as handle:
data = json.load(handle)
self._partners = [InfluencerPartner(**entry) for entry in data]
def match_partners(
self, tokens: Sequence[str], focus: str | None, limit: int
) -> List[Tuple[float, InfluencerPartner]]:
token_set = {token.lower() for token in tokens}
focus_tokens = {token.lower() for token in re.findall(r"[a-zA-Z0-9]+", focus or "")}
ranked: List[Tuple[float, InfluencerPartner]] = []
for partner in self._partners:
haystack = (
f"{partner.name} {partner.platform} {partner.niche} {partner.conversion_focus} "
f"{' '.join(partner.keywords)} {partner.case_study}"
).lower()
score = 0.0
keyword_hits = 0
for token in token_set:
if token in haystack:
score += 1.8
keyword_hits += 1
if token in {keyword.lower() for keyword in partner.keywords}:
score += 1.5
keyword_hits += 1
for token in focus_tokens:
if token and token in haystack:
score += 1.2
if focus and focus.lower() in partner.conversion_focus.lower():
score += 1.0
if focus and focus.lower() in partner.niche.lower():
score += 0.8
score += min(partner.engagement_rate * 20, 2.4)
score += min(partner.audience_size / 150000.0, 2.0)
score += min(keyword_hits * 0.2, 1.5)
if "ai" in token_set and "ai" in {k.lower() for k in partner.keywords}:
score += 0.6
ranked.append((score, partner))
ranked.sort(key=lambda item: item[0], reverse=True)
return ranked[: max(1, limit)]
@dataclass(slots=True)
class AmazonMarketplaceListing:
"""Amazon or AWS Marketplace listing metadata."""
asin: str
title: str
category: str
price_usd: float
commission_rate: float
rating: float
reviews: int
url: str
keywords: List[str]
estimated_monthly_sales: int
buy_box_share: float
prime_eligible: bool
fulfillment_model: str
@dataclass(slots=True)
class AmazonProductRecommendation:
"""Recommendation payload produced by the Amazon marketplace process."""
asin: str
title: str
category: str
price_usd: float
commission_rate: float
rating: float
reviews: int
estimated_monthly_sales: int
buy_box_share: float
prime_eligible: bool
fulfillment_model: str
projected_monthly_revenue: float
rationale: str
url: str
source: str
notes: str | None = None
class AmazonMarketplaceCatalog:
"""Loads curated Amazon and AWS marketplace listings for offline use."""
def __init__(self, listings: Sequence[AmazonMarketplaceListing] | None = None):
if listings is not None:
self._listings = list(listings)
else:
with open(AMAZON_MARKETPLACE_DATASET, "r", encoding="utf-8") as handle:
data = json.load(handle)
self._listings = [AmazonMarketplaceListing(**entry) for entry in data]
def match_listings(
self, tokens: Sequence[str], limit: int, focus: str | None = None
) -> List[Tuple[float, AmazonMarketplaceListing]]:
ranked: List[Tuple[float, AmazonMarketplaceListing]] = []
token_set = {token.lower() for token in tokens}
for listing in self._listings:
haystack = f"{listing.title} {listing.category} {' '.join(listing.keywords)}".lower()
score = 0.0
keyword_hits = 0
for token in token_set:
if token in haystack:
score += 1.5
keyword_hits += 1
if token in {keyword.lower() for keyword in listing.keywords}:
score += 1.8
if focus:
if focus.lower() in listing.category.lower():
score += 1.2
if focus.lower() in haystack:
score += 0.8
score += min(listing.rating / 5.0, 1.0)
score += min(listing.reviews / 500.0, 1.0)
score += min(listing.estimated_monthly_sales / 400.0, 1.2)
score += min(listing.buy_box_share, 1.0)
if listing.prime_eligible:
score += 0.35
if listing.fulfillment_model.lower().startswith("fba"):
score += 0.25
score += keyword_hits * 0.1
ranked.append((score, listing))
ranked.sort(key=lambda item: item[0], reverse=True)
return ranked[: max(1, limit)]
class AmazonMarketplaceProcess:
"""Transforms marketplace listings into revenue plans with optional SDK enrichment."""
REQUIRED_ENV_VARS = {
"AMAZON_PAAPI_ACCESS_KEY": "access_key",
"AMAZON_PAAPI_SECRET_KEY": "[credential removed]",
"AMAZON_PAAPI_PARTNER_TAG": "partner_tag",
}
OPTIONAL_ENV_VARS = {
"AWS_REGION": "region",
"AWS_PROFILE": "profile",
}
def __init__(
self,
catalog: AmazonMarketplaceCatalog | None = None,
config: Dict[str, Any] | None = None,
) -> None:
self._catalog = catalog or AmazonMarketplaceCatalog()
self._config = config or self._config_from_env()
self._session = self._build_session()
def _build_session(self) -> Any | None:
if not self.configured or boto3 is None:
return None
try: # pragma: no cover - requires boto3 dependency
return boto3.session.Session(
aws_access_key_id=self._config.get("access_key"),
aws_secret_access_key=[credential removed]"secret_key"),
region_name=self._config.get("region"),
profile_name=self._config.get("profile"),
)
except Exception:
return None
@classmethod
def _config_from_env(cls) -> Dict[str, Any]:
config: Dict[str, Any] = {}
for env_var, key in {**cls.REQUIRED_ENV_VARS, **cls.OPTIONAL_ENV_VARS}.items():
value = os.getenv(env_var)
if value:
config[key] = value
return config
@property
def configured(self) -> bool:
required = {key for key in self.REQUIRED_ENV_VARS.values()}
return required <= self._config.keys()
def listings(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int,
) -> List[AmazonProductRecommendation]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
focus = None
if campaigns:
focus = campaigns[0].campaign.channel.split()[0]
matches = self._catalog.match_listings(tokens, limit, focus)
if not matches:
return []
avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.1
avg_ctr = mean([score.metrics.expected_ctr for score in campaigns]) if campaigns else 0.025
avg_confidence = (
mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.55
)
recommendations: List[AmazonProductRecommendation] = []
for match_score, listing in matches:
keyword_overlap = sorted(
{
token
for token in tokens
if token in {kw.lower() for kw in listing.keywords}
or token in listing.title.lower()
}
)
intensity = max(0.1, min(2.0, match_score / 4.0))
volume_estimate = listing.estimated_monthly_sales * max(0.4, avg_roi / 2.5)
buy_box_lift = 0.75 + listing.buy_box_share * 0.6
prime_lift = 1.18 if listing.prime_eligible else 0.9
confidence_lift = 0.85 + avg_confidence * 0.3
intent_multiplier = 0.35 + intensity * 0.25 + avg_ctr * 9.5
projected_revenue = round(
listing.price_usd
* listing.commission_rate
* volume_estimate
* buy_box_lift
* prime_lift
* confidence_lift
* intent_multiplier
/ 100,
2,
)
if keyword_overlap:
rationale = (
f"Aligns {listing.category.lower()} offer with {', '.join(keyword_overlap)} demand signals."
)
else:
rationale = (
f"High rating and volume listing positioned for {listing.category.lower()} expansion."
)
source = "amazon_paapi" if self._session else "amazon_catalog"
notes = None
if not self._session:
notes = (
"Simulated listing intelligence; supply Amazon PA-API credentials to activate live harvesting."
)
recommendations.append(
AmazonProductRecommendation(
asin=listing.asin,
title=listing.title,
category=listing.category,
price_usd=listing.price_usd,
commission_rate=listing.commission_rate,
rating=listing.rating,
reviews=listing.reviews,
estimated_monthly_sales=listing.estimated_monthly_sales,
buy_box_share=listing.buy_box_share,
prime_eligible=listing.prime_eligible,
fulfillment_model=listing.fulfillment_model,
projected_monthly_revenue=projected_revenue,
rationale=rationale,
url=listing.url,
source=source,
notes=notes,
)
)
return recommendations
@dataclass(slots=True)
class EbayMarketplaceListing:
"""eBay marketplace listing metadata for revenue planning."""
item_id: str
title: str
category: str
price_usd: float
affiliate_rate: float
seller_level: str
seller_rating: float
conversion_rate: float
sold_per_month: int
shipping_days: int
url: str
keywords: List[str]
@dataclass(slots=True)
class EbayRevenuePlay:
"""Recommendation payload produced by the eBay marketplace process."""
item_id: str
title: str
category: str
price_usd: float
affiliate_rate: float
seller_level: str
seller_rating: float
conversion_rate: float
sold_per_month: int
projected_monthly_revenue: float
projected_profit_margin: float
shipping_days: int
rationale: str
url: str
source: str
notes: str | None = None
class EbayMarketplaceCatalog:
"""Loads curated eBay marketplace listings for offline use."""
def __init__(self, listings: Sequence[EbayMarketplaceListing] | None = None):
if listings is not None:
self._listings = list(listings)
else:
with open(EBAY_MARKETPLACE_DATASET, "r", encoding="utf-8") as handle:
data = json.load(handle)
self._listings = [EbayMarketplaceListing(**entry) for entry in data]
def match_listings(
self, tokens: Sequence[str], limit: int, focus: str | None = None
) -> List[Tuple[float, EbayMarketplaceListing]]:
ranked: List[Tuple[float, EbayMarketplaceListing]] = []
token_set = {token.lower() for token in tokens}
for listing in self._listings:
haystack = (
f"{listing.title} {listing.category} {' '.join(listing.keywords)}"
).lower()
score = 0.0
keyword_hits = 0
for token in token_set:
if token in haystack:
score += 1.4
keyword_hits += 1
if token in {keyword.lower() for keyword in listing.keywords}:
score += 1.6
if focus:
focus_lower = focus.lower()
if focus_lower in listing.category.lower():
score += 1.0
if focus_lower in haystack:
score += 0.6
score += min(listing.seller_rating / 5.0, 1.0)
score += min(listing.conversion_rate * 80.0, 1.2)
score += min(listing.sold_per_month / 350.0, 1.2)
if listing.seller_level.lower().startswith("top"):
score += 0.35
if listing.shipping_days <= 3:
score += 0.25
score += keyword_hits * 0.1
ranked.append((score, listing))
ranked.sort(key=lambda item: item[0], reverse=True)
return ranked[: max(1, limit)]
class EbayMarketplaceProcess:
"""Transforms eBay inventory into performance-driven affiliate plays."""
REQUIRED_ENV_VARS = {
"EBAY_APP_ID": "app_id",
"EBAY_CERT_ID": "cert_id",
"EBAY_DEV_ID": "dev_id",
}
OPTIONAL_ENV_VARS = {
"EBAY_REDIRECT_URI": "redirect_uri",
"EBAY_CAMPAIGN_ID": "campaign_id",
}
def __init__(
self,
catalog: EbayMarketplaceCatalog | None = None,
config: Dict[str, Any] | None = None,
) -> None:
self._catalog = catalog or EbayMarketplaceCatalog()
self._config = config or self._config_from_env()
self._session = self._build_session()
def _build_session(self) -> Any | None:
if not self.configured or requests is None:
return None
try:
session = requests.Session()
session.headers.update({"User-Agent": "ai-brain-marketing-ani/1.0"})
return session
except Exception:
return None
@classmethod
def _config_from_env(cls) -> Dict[str, Any]:
config: Dict[str, Any] = {}
for env_var, key in {**cls.REQUIRED_ENV_VARS, **cls.OPTIONAL_ENV_VARS}.items():
value = os.getenv(env_var)
if value:
config[key] = value
return config
@property
def configured(self) -> bool:
required = {key for key in self.REQUIRED_ENV_VARS.values()}
return required <= self._config.keys()
def listings(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int,
) -> List[EbayRevenuePlay]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
focus = None
if campaigns:
focus = campaigns[0].campaign.channel.split()[0]
matches = self._catalog.match_listings(tokens, limit, focus)
if not matches:
return []
avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.0
avg_confidence = (
mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.5
)
recommendations: List[EbayRevenuePlay] = []
for match_score, listing in matches:
keyword_overlap = sorted(
{
token
for token in tokens
if token in {kw.lower() for kw in listing.keywords}
or token in listing.title.lower()
}
)
intensity = max(0.2, min(2.2, match_score / 3.8))
velocity = listing.sold_per_month * max(0.45, avg_roi / 2.2)
conversion_lift = 0.6 + listing.conversion_rate * 14.0
seller_lift = 0.8 + (listing.seller_rating / 5.0) * 0.4
speed_lift = 1.05 + max(0, (4 - listing.shipping_days)) * 0.04
confidence_lift = 0.8 + avg_confidence * 0.35
projected_revenue = round(
listing.price_usd
* listing.affiliate_rate
* velocity
* conversion_lift
* seller_lift
* speed_lift
* confidence_lift
* intensity
/ 100,
2,
)
projected_margin = round(
min(0.92, listing.affiliate_rate * (conversion_lift * 0.6 + 0.35)),
2,
)
if keyword_overlap:
rationale = (
f"Targets {', '.join(keyword_overlap)} buyers with proven conversion velocity."
)
else:
rationale = (
f"Top-rated {listing.category.lower()} seller with accelerated sell-through momentum."
)
source = "ebay_partner_api" if self._session else "ebay_catalog"
notes = None
if not self._session:
notes = (
"Offline catalog heuristics; supply eBay developer credentials to activate live sourcing."
)
recommendations.append(
EbayRevenuePlay(
item_id=listing.item_id,
title=listing.title,
category=listing.category,
price_usd=listing.price_usd,
affiliate_rate=listing.affiliate_rate,
seller_level=listing.seller_level,
seller_rating=listing.seller_rating,
conversion_rate=listing.conversion_rate,
sold_per_month=listing.sold_per_month,
projected_monthly_revenue=projected_revenue,
projected_profit_margin=projected_margin,
shipping_days=listing.shipping_days,
rationale=rationale,
url=listing.url,
source=source,
notes=notes,
)
)
return recommendations
@dataclass(slots=True)
class CryptoMarketSnapshot:
"""Market intelligence payload for a cryptocurrency asset."""
asset: str
price_usd: float | None
change_24h: float | None
market_cap: float | None
volume_24h: float | None
source: str
notes: str | None = None
@dataclass(slots=True)
class StockMarketRecord:
"""Static fundamentals for a global stock market or macro basket."""
symbol: str
name: str
region: str
sector_focus: str
current_price: float
avg_daily_volume: float
market_cap: float
pe_ratio: float
dividend_yield: float
volatility_30d: float
momentum_score: float
macro_signal: str
marketing_correlation: float
crypto_correlation: float | None = None
@dataclass(slots=True)
class StockMarketForecast:
"""Forecast intelligence summarizing statistical outlooks for a market."""
symbol: str
name: str
region: str
focus: str
current_price: float
trend: str
forecast_return_30d: float
forecast_return_90d: float
forecast_revenue_multiplier: float
volatility: float
confidence: float
signal_summary: str
source: str
notes: str | None = None
class CryptoMarketProcess:
"""Ingests crypto market data across public APIs with graceful fallbacks."""
ASSET_ALIASES: Dict[str, str] = {
"btc": "bitcoin",
"bitcoin": "bitcoin",
"sats": "bitcoin",
"eth": "ethereum",
"ethereum": "ethereum",
"sol": "solana",
"solana": "solana",
"matic": "polygon",
"polygon": "polygon",
"doge": "dogecoin",
"dogecoin": "dogecoin",
}
DEFAULT_ASSETS: List[str] = ["bitcoin", "ethereum", "solana"]
FALLBACK_PRICES: Dict[str, Dict[str, Any]] = {
"bitcoin": {
"price_usd": 68000.0,
"change_24h": 2.5,
"market_cap": 1.3e12,
"volume_24h": 3.1e10,
"source": "simulated",
"notes": "Use CoinGecko or CoinCap credentials for live BTC spot feeds.",
},
"ethereum": {
"price_usd": 3500.0,
"change_24h": 3.1,
"market_cap": 4.2e11,
"volume_24h": 1.8e10,
"source": "simulated",
"notes": "Hook CoinGecko or CoinCap to mirror ETH market structure.",
},
"solana": {
"price_usd": 160.0,
"change_24h": 4.8,
"market_cap": 7.2e10,
"volume_24h": 4.1e9,
"source": "simulated",
"notes": "Pair CoinGecko streams for SOL liquidity snapshots.",
},
"polygon": {
"price_usd": 0.95,
"change_24h": 1.2,
"market_cap": 8.8e9,
"volume_24h": 6.7e8,
"source": "simulated",
"notes": "Integrate CoinCap assets endpoint for live Polygon metrics.",
},
"dogecoin": {
"price_usd": 0.18,
"change_24h": 5.5,
"market_cap": 2.6e10,
"volume_24h": 9.2e8,
"source": "simulated",
"notes": "CoinGecko provides DOGE spot pricing with rate-limited tier.",
},
}
def __init__(
self,
session: Any | None = None,
timeout: float = 4.0,
allow_network: bool = True,
) -> None:
self._session = None
if allow_network and requests is not None:
self._session = session or requests.Session()
self._timeout = timeout
@property
def configured(self) -> bool:
return self._session is not None
def _normalize_asset(self, token: str) -> str | None:
return self.ASSET_ALIASES.get(token.lower())
def _infer_assets(self, prompt: str, limit: int) -> List[str]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
assets: List[str] = []
for token in tokens:
asset = self._normalize_asset(token)
if asset and asset not in assets:
assets.append(asset)
for asset in self.DEFAULT_ASSETS:
if asset not in assets:
assets.append(asset)
return assets[: max(1, limit)]
def _request_json(
self, url: str, params: Dict[str, Any] | None = None
) -> Dict[str, Any] | List[Any] | None:
if not self._session:
return None
try: # pragma: no cover - requires network
response = self._session.get(url, params=params, timeout=self._timeout)
response.raise_for_status()
return response.json()
except Exception:
return None
def _fetch_coingecko(self, asset: str) -> Dict[str, Any] | None:
data = self._request_json(
"https://api.coingecko.com/api/v3/simple/price",
params={
"ids": asset,
"vs_currencies": "usd",
"include_24hr_change": "true",
"include_market_cap": "true",
"include_24hr_vol": "true",
},
)
if isinstance(data, dict) and asset in data:
entry = data[asset]
return {
"price_usd": float(entry.get("usd")) if entry.get("usd") is not None else None,
"change_24h": float(entry.get("usd_24h_change")) if entry.get("usd_24h_change") is not None else None,
"market_cap": float(entry.get("usd_market_cap")) if entry.get("usd_market_cap") is not None else None,
"volume_24h": float(entry.get("usd_24h_vol")) if entry.get("usd_24h_vol") is not None else None,
"source": "coingecko",
}
return None
def _fetch_coincap(self, asset: str) -> Dict[str, Any] | None:
data = self._request_json(f"https://api.coincap.io/v2/assets/{asset}")
if isinstance(data, dict) and isinstance(data.get("data"), dict):
entry = data["data"]
try:
price = float(entry.get("priceUsd")) if entry.get("priceUsd") is not None else None
change = float(entry.get("changePercent24Hr")) if entry.get("changePercent24Hr") is not None else None
market_cap = float(entry.get("marketCapUsd")) if entry.get("marketCapUsd") is not None else None
volume = float(entry.get("volumeUsd24Hr")) if entry.get("volumeUsd24Hr") is not None else None
except (TypeError, ValueError):
price = change = market_cap = volume = None
return {
"price_usd": price,
"change_24h": change,
"market_cap": market_cap,
"volume_24h": volume,
"source": "coincap",
}
return None
def _fetch_blockstream(self) -> str | None:
data = self._request_json("https://blockstream.info/api/mempool")
if isinstance(data, dict) and {"count", "vsize"} <= data.keys():
return f"Blockstream mempool: {data['count']} txs, {data['vsize']} vsize"
return None
def _fetch_blockcypher(self, asset: str) -> str | None:
if asset != "bitcoin":
return None
data = self._request_json("https://api.blockcypher.com/v1/btc/main")
if isinstance(data, dict) and {"height", "unconfirmed_count"} <= data.keys():