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 ↓10,463 lines · 396.6 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
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)
total = optimistic_prob + base_prob + conservative_prob
optimistic_prob = round(optimistic_prob / total, 2)
base_prob = round(base_prob / total, 2)
conservative_prob = round(conservative_prob / total, 2)
accelerators = self._signal_accelerators(
google_ads, cj_affiliate, amazon, ebay, aws, stripe, paypal, influencers
)
stabilizers = self._stabilizers(diagnostics, budget)
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"
)
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,
capabilities: Sequence[PhysicsEconomicsCapability] | None = None,
dataset_path: str = PHYSICS_ECONOMICS_DATASET,
config: Dict[str, Any] | None = None,
) -> None:
self._capabilities = (
list(capabilities)
if capabilities is not None
else self._load(dataset_path)
)
self._config = config or self._config_from_env()
@classmethod
def _config_from_env(cls) -> Dict[str, List[str]]:
config: Dict[str, set[str]] = {}
for env_var, key in cls.OPTIONAL_ENV_VARS.items():
if os.getenv(env_var):
config.setdefault(key, set()).add(env_var.lower())
return {key: sorted(values) for key, values in config.items()}
@property
def configured(self) -> bool:
return bool(self._config)
@staticmethod
def _load(path: str) -> List[PhysicsEconomicsCapability]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
catalog: List[PhysicsEconomicsCapability] = []
if isinstance(payload, list):
for entry in payload:
if isinstance(entry, dict):
try:
catalog.append(PhysicsEconomicsCapability.from_dict(entry))
except (KeyError, TypeError, ValueError):
continue
if not catalog:
catalog.append(
PhysicsEconomicsCapability(
name="Foundational Physics & Economics Readiness",
domain="physics_economics",
description="Fallback capability that maps marketing prompts to basic physics and economic guardrails.",
trigger_keywords=["physics", "economics", "quantum", "pricing", "compression"],
marketing_algorithms=["Baseline ROI sensitivity analysis"],
physics_principles=["System dynamics balancing"],
quantum_strategies=["Monte Carlo variance reduction"],
economic_models=["Unit economics calculator"],
hardware_strategies=["CPU efficient inference"],
data_compression=["Vector sketching"],
risk_controls=["Stress test budgets weekly"],
metrics=["roi_guardrail"],
maturity_signals=["Fallback heuristics"],
base_maturity="emerging",
orchestration_play="Use fallback heuristics to align marketing spend with baseline physics/economics guardrails.",
docs=[],
)
)
return catalog
@staticmethod
def _tokenize(text: str) -> List[str]:
return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2]
@staticmethod
def _phrase_hits(collection: Sequence[str], tokens: Sequence[str]) -> List[str]:
hits: List[str] = []
token_set = set(tokens)
for phrase in collection:
normalized = phrase.lower()
if any(token in normalized for token in token_set):
hits.append(phrase)
return hits
def plans(
self,
prompt: str,
campaigns: Sequence["CampaignScore"],
limit: int = 3,
) -> List[PhysicsEconomicsPlan]:
if not campaigns:
raise ValueError("Physics economics plans require campaign scores")
prompt_tokens = set(self._tokenize(prompt))
campaign_tokens = set(
token
for campaign in campaigns
for token in self._tokenize(
" ".join(
[
campaign.campaign.channel,
campaign.campaign.audience,
campaign.campaign.offer,
campaign.campaign.objective,
]
)
)
)
avg_roi = mean(score.metrics.expected_roi for score in campaigns)
avg_confidence = mean(score.metrics.confidence for score in campaigns)
plans: List[Tuple[float, PhysicsEconomicsPlan]] = []
for capability in self._capabilities:
keyword_hits = [
kw
for kw in capability.trigger_keywords
if kw in prompt_tokens or kw in campaign_tokens
]
physics_hits = self._phrase_hits(capability.physics_principles, prompt_tokens)
quantum_hits = self._phrase_hits(capability.quantum_strategies, prompt_tokens)
economics_hits = self._phrase_hits(capability.economic_models, prompt_tokens)
hardware_hits = self._phrase_hits(capability.hardware_strategies, prompt_tokens)
compression_hits = self._phrase_hits(capability.data_compression, prompt_tokens)
base_score = 0.45 if capability.base_maturity in {"advanced", "elite"} else 0.35
score = base_score
score += 0.07 * len(keyword_hits)
score += 0.05 * len(physics_hits)
score += 0.05 * len(quantum_hits)
score += 0.06 * len(economics_hits)
score += 0.04 * len(hardware_hits)
score += 0.03 * len(compression_hits)
if avg_roi >= 2.5:
score += 0.05
if avg_confidence >= 0.65:
score += 0.04
if {"quantum", "physics", "economics", "pricing"} & prompt_tokens:
score += 0.05
if self.configured:
score += 0.08
maturity_levels = ["emerging", "advanced", "elite", "autonomous"]
base_index = maturity_levels.index(capability.base_maturity)
if score >= 1.15:
maturity_index = min(base_index + 2, len(maturity_levels) - 1)
elif score >= 0.85:
maturity_index = min(base_index + 1, len(maturity_levels) - 1)
else:
maturity_index = base_index
maturity = maturity_levels[maturity_index]
reasoning_parts: List[str] = []
if keyword_hits:
reasoning_parts.append(
f"Prompt references {', '.join(keyword_hits[:3])} indicating physics/economics urgency."
)
if economics_hits:
reasoning_parts.append(
f"Economic models {', '.join(economics_hits[:2])} align with revenue objectives."
)
if physics_hits or quantum_hits:
physics_blend = physics_hits[:2] + quantum_hits[:1]
if physics_blend:
reasoning_parts.append(
f"Physics stack engages {', '.join(physics_blend)} for stress stability."
)
if not reasoning_parts:
reasoning_parts.append(
f"Campaign ROI {avg_roi:.2f}x and confidence {avg_confidence:.0%} justify activating {capability.name}."
)
reasoning = " ".join(reasoning_parts)
orchestration_steps: List[str] = [capability.orchestration_play]
if capability.marketing_algorithms:
orchestration_steps.append(
"Algorithms: " + ", ".join(capability.marketing_algorithms[:3])
)
if capability.physics_principles:
orchestration_steps.append(
"Physics focus: " + ", ".join(capability.physics_principles[:2])
)
if capability.economic_models:
orchestration_steps.append(
"Economic models: " + ", ".join(capability.economic_models[:2])
)
if capability.hardware_strategies:
orchestration_steps.append(
"Hardware plan: " + ", ".join(capability.hardware_strategies[:2])
)
if capability.data_compression:
orchestration_steps.append(
"Compression: " + ", ".join(capability.data_compression[:2])
)
if capability.risk_controls:
orchestration_steps.append(
"Risk controls: " + ", ".join(capability.risk_controls[:2])
)
source = "physics_economics_live" if self.configured else "physics_economics_catalog"
notes: str | None = None
if not self.configured:
notes = (
"Simulated from physics/economics catalog; set QISKIT_RUNTIME / AWS_BRAKET_REGION / ECONOMICS_DATA_WAREHOUSE "
"to enable live telemetry."
)
plans.append(
(
score,
PhysicsEconomicsPlan(
capability=capability.name,
domain=capability.domain,
maturity=maturity,
description=capability.description,
rationale=reasoning,
marketing_algorithms=list(capability.marketing_algorithms),
physics_principles=list(capability.physics_principles),
quantum_strategies=list(capability.quantum_strategies),
economic_models=list(capability.economic_models),
hardware_strategies=list(capability.hardware_strategies),
data_compression=list(capability.data_compression),
risk_controls=list(capability.risk_controls),
metrics=list(capability.metrics),
orchestration_steps=orchestration_steps,
docs=list(capability.docs),
source=source,
notes=notes,
),
)
)
plans.sort(key=lambda item: item[0], reverse=True)
return [plan for _, plan in plans[: max(1, limit)]]
class PaymentIntegrationProcess:
"""Plans how the ANI orchestrates payment provider integrations."""
PROVIDER_ENV_HINTS: Dict[str, List[str]] = {
"stripe": ["STRIPE_API_KEY", "STRIPE_ACCOUNT_ID"],
"paypal": ["PAYPAL_CLIENT_ID", "PAYPAL_CLIENT_SECRET", "PAYPAL_MERCHANT_ID"],
"cash app": ["SQUARE_ACCESS_TOKEN", "SQUARE_APPLICATION_ID", "SQUARE_LOCATION_ID"],
"venmo": ["VENMO_CLIENT_ID", "VENMO_CLIENT_SECRET", "VENMO_ACCESS_TOKEN"],
}
def __init__(
self,
capabilities: Sequence[PaymentPlatformCapability] | None = None,
dataset_path: str = PAYMENT_PLATFORMS_DATASET,
config: Dict[str, List[str]] | None = None,
) -> None:
self._capabilities = (
list(capabilities)
if capabilities is not None
else self._load(dataset_path)
)
self._config = config or self._config_from_env()
@classmethod
def _config_from_env(cls) -> Dict[str, List[str]]:
config: Dict[str, List[str]] = {}
for provider_key, env_vars in cls.PROVIDER_ENV_HINTS.items():
available = [env for env in env_vars if os.getenv(env)]
if available:
config[provider_key] = sorted(set(env.lower() for env in available))
return config
@property
def configured_providers(self) -> set[str]:
return set(self._config.keys())
@staticmethod
def _load(path: str) -> List[PaymentPlatformCapability]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
catalog: List[PaymentPlatformCapability] = []
if isinstance(payload, list):
for entry in payload:
if isinstance(entry, dict):
try:
catalog.append(PaymentPlatformCapability.from_dict(entry))
except (KeyError, TypeError, ValueError):
continue
if not catalog:
catalog.append(
PaymentPlatformCapability(
provider="Stripe",
description="Fallback Stripe guidance for payment readiness.",
integration_focus=["Checkout"],
docs=["https://docs.stripe.com/"],
sdks=["stripe"],
setup=["Review Stripe docs"],
troubleshooting=["Verify API keys"],
codefix=["Regenerate secret keys"],
marketing_plays=["Launch payment link"],
keywords=["stripe"],
base_maturity="foundational",
)
)
return catalog
@staticmethod
def _tokenize(text: str) -> List[str]:
return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2]
def plans(
self,
prompt: str,
campaigns: Sequence["CampaignScore"],
limit: int = 4,
) -> List[PaymentIntegrationPlan]:
if not campaigns:
raise ValueError("Payment integration plans require campaign scores")
prompt_tokens = set(self._tokenize(prompt))
campaign_tokens = set(
token
for campaign in campaigns
for token in self._tokenize(
" ".join(
[
campaign.campaign.channel,
campaign.campaign.audience,
campaign.campaign.offer,
campaign.campaign.objective,
]
)
)
)
plans: List[Tuple[int, PaymentIntegrationPlan]] = []
for capability in self._capabilities:
keywords = set(capability.keywords)
prompt_hits = sorted(prompt_tokens & keywords)
campaign_hits = sorted(campaign_tokens & keywords)
score = len(prompt_hits) * 2 + len(campaign_hits)
if any(key in prompt.lower() for key in capability.keywords):
score += 1
provider_key = capability.provider.lower()
configured = False
for key, envs in self._config.items():
if provider_key.startswith(key):
configured = True
provider_key = key
break
if configured and score >= 3:
maturity = "autonomous"
elif score >= 3:
maturity = "advanced"
elif capability.base_maturity:
maturity = capability.base_maturity
else:
maturity = "foundational"
reasoning_parts = []
if prompt_hits:
reasoning_parts.append(
f"Prompt references {', '.join(prompt_hits[:3])} signals"
)
if campaign_hits:
reasoning_parts.append(
f"Campaign corpus echoes {', '.join(campaign_hits[:3])} demand"
)
if configured:
envs = ", ".join(self._config.get(provider_key, []))
reasoning_parts.append(f"Environment exposes {envs} credentials")
if not reasoning_parts:
reasoning_parts.append("Fallback payment integration guidance available")
notes = None
if not configured:
hint_envs = self.PROVIDER_ENV_HINTS.get(provider_key, [])
if hint_envs:
notes = "Export env vars: " + ", ".join(hint_envs)
plans.append(
(
score,
PaymentIntegrationPlan(
provider=capability.provider,
maturity=maturity,
description=capability.description,
reasoning="; ".join(reasoning_parts) + ".",
integration_focus=list(capability.integration_focus),
docs=list(capability.docs),
sdks=list(capability.sdks),
setup=list(capability.setup),
troubleshooting=list(capability.troubleshooting),
codefix=list(capability.codefix),
marketing_plays=list(capability.marketing_plays),
source=capability.source,
notes=notes,
),
)
)
plans.sort(key=lambda item: item[0], reverse=True)
return [plan for _, plan in plans[: max(1, limit)]]
class InvestingProcess:
"""Designs investing platform activation plans."""
def __init__(
self,
capabilities: Sequence[InvestingCapability] | None = None,
dataset_path: str = INVESTING_PLATFORMS_DATASET,
) -> None:
self._capabilities = (
list(capabilities)
if capabilities is not None
else self._load(dataset_path)
)
@staticmethod
def _load(path: str) -> List[InvestingCapability]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
catalog: List[InvestingCapability] = []
if isinstance(payload, list):
for entry in payload:
if isinstance(entry, dict):
try:
catalog.append(InvestingCapability.from_dict(entry))
except (KeyError, TypeError, ValueError):
continue
if not catalog:
catalog.append(
InvestingCapability(
provider="Fallback Investing",
category="investing",
description="Baseline investing plan to highlight brokerage monetization paths.",
asset_classes=["equities", "etfs"],
onboarding=["Publish investing landing page", "Launch nurture emails"],
monetization=["Promote brokerage affiliate", "Sell premium research"],
risk_controls=["Disclose investing risks", "Monitor compliance copy"],
tools=["Broker dashboard"],
keywords=["invest", "brokerage", "stock"],
revenue_programs=["Broker affiliate"],
kpis=["funded_accounts", "aum_contributed"],
base_maturity="foundational",
source="investing_platforms_fallback",
)
)
return catalog
@staticmethod
def _tokenize(text: str) -> List[str]:
return [token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2]
@staticmethod
def _phrase_hits(collection: Sequence[str], token_set: set[str]) -> List[str]:
hits: List[str] = []
for phrase in collection:
normalized = phrase.lower()
if any(token in normalized for token in token_set):
hits.append(phrase)
return hits
def plans(
self,
prompt: str,
campaigns: Sequence["CampaignScore"],
limit: int = 3,
) -> List[InvestingPlan]:
if not campaigns:
raise ValueError("Investing plans require campaign scores")
prompt_tokens = set(self._tokenize(prompt))
campaign_tokens = set(
token
for campaign in campaigns
for token in self._tokenize(
" ".join(
[
campaign.campaign.channel,
campaign.campaign.offer,
campaign.campaign.objective,
campaign.campaign.audience,
]
)
)
)
combined_tokens = prompt_tokens | campaign_tokens
avg_roi = mean(score.metrics.expected_roi for score in campaigns)
avg_confidence = mean(score.metrics.confidence for score in campaigns)
plans: List[Tuple[float, InvestingPlan]] = []
for capability in self._capabilities:
keyword_hits = sorted(set(capability.keywords) & combined_tokens)
asset_hits = self._phrase_hits(capability.asset_classes, combined_tokens)
monetization_hits = self._phrase_hits(capability.monetization, combined_tokens)
onboarding_hits = self._phrase_hits(capability.onboarding, prompt_tokens)
category_hit = capability.category.lower() in combined_tokens
score = 0.45
score += 0.1 * len(keyword_hits)
score += 0.07 * len(asset_hits)
score += 0.06 * len(monetization_hits)
score += 0.05 * len(onboarding_hits)
if category_hit:
score += 0.08
if "invest" in prompt_tokens or "trading" in prompt_tokens:
score += 0.05
if avg_roi >= 2.5:
score += 0.06
if avg_confidence >= 0.65:
score += 0.04
if score >= 1.05:
maturity = "autonomous"
elif score >= 0.8:
maturity = "advanced"
else:
maturity = "foundational"
base_order = ["foundational", "advanced", "autonomous"]
if base_order.index(maturity) < base_order.index(capability.base_maturity):
maturity = capability.base_maturity
reasoning_parts: List[str] = []
if keyword_hits:
reasoning_parts.append(
f"Prompt and campaigns reference {', '.join(keyword_hits[:3])} investing signals"
)
if asset_hits:
reasoning_parts.append(
f"Asset focus aligns with {', '.join(asset_hits[:3])} plays"
)
if monetization_hits:
reasoning_parts.append(
f"Monetization hooks mention {', '.join(monetization_hits[:2])}"
)
if onboarding_hits:
reasoning_parts.append(
f"Prompt language maps to onboarding step {onboarding_hits[0]}"
)
if not reasoning_parts:
reasoning_parts.append("Fallback investing monetization playbook ready")
plans.append(
(
score,
InvestingPlan(
provider=capability.provider,
category=capability.category,
maturity=maturity,
description=capability.description,
reasoning="; ".join(reasoning_parts) + ".",
asset_classes=list(capability.asset_classes),
monetization=list(capability.monetization),
onboarding=list(capability.onboarding),
tools=list(capability.tools),
risk_controls=list(capability.risk_controls),
revenue_programs=list(capability.revenue_programs),
kpis=list(capability.kpis),
source=capability.source,
notes=None,
),
)
)
plans.sort(key=lambda item: item[0], reverse=True)
return [plan for _, plan in plans[: max(1, limit)]]
class KnowledgeWorkflowProcess:
"""Designs retrieval and knowledge graph plans for the ANI."""
def __init__(
self,
assets: Sequence[KnowledgeWorkflowAsset] | None = None,
dataset_path: str = KNOWLEDGE_WORKFLOWS_DATASET,
) -> None:
self._assets = list(assets) if assets is not None else self._load(dataset_path)
@staticmethod
def _load(path: str) -> List[KnowledgeWorkflowAsset]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
catalog: List[KnowledgeWorkflowAsset] = []
if isinstance(payload, list):
for entry in payload:
if isinstance(entry, dict):
try:
catalog.append(KnowledgeWorkflowAsset.from_dict(entry))
except (KeyError, TypeError, ValueError):
continue
if not catalog:
catalog.append(
KnowledgeWorkflowAsset(
name="Foundational Retrieval Workflow",
category="rag",
description="Fallback retrieval workflow for campaign collateral.",
trigger_keywords=["campaign", "retrieval"],
indexing_steps=["Collect campaign documents", "Chunk and embed"],
retrieval_steps=["Similarity search", "Augment prompt"],
vector_store="faiss",
embedding_models=["sentence-transformers/all-MiniLM-L6-v2"],
tooling=["LangChain"],
evaluation=["context_precision"],
governance=["document versioning"],
caching=["prompt cache"],
source="knowledge_workflows_fallback",
)
)
return catalog
@staticmethod
def _tokenize(text: str) -> set[str]:
return {token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2}
def plans(
self,
prompt: str,
campaigns: Sequence["CampaignScore"],
limit: int = 3,
) -> List[KnowledgeWorkflowPlan]:
if not campaigns:
raise ValueError("Knowledge workflow plans require campaign scores")
prompt_tokens = self._tokenize(prompt)
campaign_corpus = " ".join(score.campaign.raw for score in campaigns)
campaign_tokens = self._tokenize(campaign_corpus)
avg_confidence = mean(score.metrics.confidence for score in campaigns)
avg_roi = mean(score.metrics.expected_roi for score in campaigns)
plans_by_category: Dict[str, List[Tuple[float, KnowledgeWorkflowPlan]]] = {}
for asset in self._assets:
keyword_hits = [kw for kw in asset.trigger_keywords if kw in prompt_tokens]
campaign_hits = [kw for kw in asset.trigger_keywords if kw in campaign_tokens]
tooling_hits = [tool for tool in asset.tooling if tool.lower() in prompt_tokens]
score = 0.45 + 0.12 * len(keyword_hits) + 0.08 * len(campaign_hits) + 0.05 * len(tooling_hits)
if asset.category == "rag" and {"rag", "retrieval", "cache"} & prompt_tokens:
score += 0.15
if asset.category == "kag" and {"kag", "graph", "knowledge"} & prompt_tokens:
score += 0.18
if asset.category == "kag" and avg_confidence >= 0.7:
score += 0.07
if asset.category == "rag" and avg_roi >= 2.5:
score += 0.07
maturity: str
if score >= 1.15:
maturity = "autonomous"
elif score >= 0.85:
maturity = "advanced"
else:
maturity = "foundational"
reasoning_parts: List[str] = []
if keyword_hits:
reasoning_parts.append(
f"Prompt signals {', '.join(keyword_hits[:3])} which align to {asset.name}."
)
if tooling_hits:
reasoning_parts.append(
f"Tooling cues {', '.join(tooling_hits[:2])} are already referenced in the brief."
)
if not reasoning_parts:
reasoning_parts.append(
f"Campaign ROI {avg_roi:.2f}x and confidence {avg_confidence:.0%} benefit from {asset.category.upper()} orchestration."
)
reasoning = " ".join(reasoning_parts)
indexing_plan = list(asset.indexing_steps)
retrieval_plan = list(asset.retrieval_steps)
primary_campaign = campaigns[0].campaign if campaigns else None
if primary_campaign:
indexing_plan.append(
f"Tag chunks with channel={primary_campaign.channel.lower()} and audience={primary_campaign.audience.lower()}"
)
retrieval_plan.append(
f"Prioritize offers related to {primary_campaign.offer.lower()} during similarity search"
)
plan = KnowledgeWorkflowPlan(
name=asset.name,
category=asset.category,
maturity=maturity,
description=asset.description,
reasoning=reasoning,
indexing_plan=indexing_plan,
retrieval_plan=retrieval_plan,
vector_store=asset.vector_store,
embedding_models=list(asset.embedding_models),
tooling=list(asset.tooling),
evaluation=list(asset.evaluation),
governance=list(asset.governance),
caching=list(asset.caching),
source=asset.source,
)
plans_by_category.setdefault(asset.category, []).append((score, plan))
plans: List[KnowledgeWorkflowPlan] = []
for category, items in plans_by_category.items():
items.sort(key=lambda item: item[0], reverse=True)
category_limit = max(1, min(limit, len(items)))
plans.extend(plan for _, plan in items[:category_limit])
return plans
class MixtureOfExpertsProcess:
"""Routes prompts through marketing monetization experts."""
def __init__(
self,
experts: Sequence[ExpertProfile] | None = None,
dataset_path: str = MOE_EXPERTS_DATASET,
) -> None:
self._experts = list(experts) if experts is not None else self._load(dataset_path)
@staticmethod
def _load(path: str) -> List[ExpertProfile]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
profiles: List[ExpertProfile] = []
if isinstance(payload, list):
for entry in payload:
if isinstance(entry, dict):
try:
profiles.append(ExpertProfile.from_dict(entry))
except (KeyError, TypeError, ValueError):
continue
if not profiles:
profiles.append(
ExpertProfile(
name="Benchmark Revenue Expert",
provider="benchmark",
description="Fallback expert using benchmark datasets to inform routing.",
trigger_keywords=["revenue", "marketing"],
specialties=["revenue operations"],
routing_signals=["roi"],
outputs=["recommendations"],
fallback="Use benchmark dataset",
metrics=["roi", "ctr"],
source="moe_experts_fallback",
)
)
return profiles
@staticmethod
def _tokenize(text: str) -> set[str]:
return {token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2}
def activate(
self,
prompt: str,
campaigns: Sequence["CampaignScore"],
limit: int = 3,
) -> List[MixtureOfExpertsPlan]:
if not campaigns:
raise ValueError("Mixture-of-experts activation requires campaign scores")
prompt_tokens = self._tokenize(prompt)
campaign_corpus = " ".join(score.campaign.raw for score in campaigns)
campaign_tokens = self._tokenize(campaign_corpus)
avg_roi = mean(score.metrics.expected_roi for score in campaigns)
avg_confidence = mean(score.metrics.confidence for score in campaigns)
scored: List[Tuple[float, ExpertProfile, float]] = []
for expert in self._experts:
keyword_hits = [kw for kw in expert.trigger_keywords if kw in prompt_tokens or kw in campaign_tokens]
specialty_hits = [
specialty
for specialty in expert.specialties
if any(word in campaign_corpus.lower() for word in specialty.lower().split())
]
provider_bonus = 0.0
if expert.provider == "google" and {"google", "sem", "ads"} & prompt_tokens:
provider_bonus += 0.18
if expert.provider == "stripe" and {"stripe", "pricing", "checkout"} & prompt_tokens: