Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/AI_Brain / services/seeder_api/intelligence.py |
| Date | Not specified in source |
| Repository | Janady13/AI_Brain |
| Revision | 840945d70b88674d8f40d330001cc63d871dc8e2 |
| Branch | codex/build-online-github-ani-nbgsix |
| Path or archive member | services/seeder_api/intelligence.py |
| Original SHA-256 | 93205dd558e063ae7437ea98254ca26dbd4cb187043c749791c6f95a7d3bf53e |
| Git blob SHA | 10a5f51b79bf9a81338775f891510fcdecfae911 |
| Reading-copy SHA-256 | 016a3ed8d5509ddef5727b148fae246d563dd34b0ba9b3ab6a509b323bedabee |
Recovered source transcript
Download complete source text ↓1,812 lines · 67.2 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
"""Marketing intelligence engine that ranks and scores enumerated campaigns."""
from __future__ import annotations
import base64
import importlib
import json
import os
import re
from dataclasses import dataclass
from functools import lru_cache
from statistics import mean
from typing import Any, Dict, List, Sequence, Tuple
try: # pragma: no cover - optional dependency that may not be installed in tests
from google.ads.googleads.client import GoogleAdsClient # type: ignore
except ImportError: # pragma: no cover - gracefully handled when SDK missing
GoogleAdsClient = None # type: ignore
try: # pragma: no cover - optional dependency that may not be installed in tests
from google.oauth2 import service_account # type: ignore
except ImportError: # pragma: no cover - gracefully handled when SDK missing
service_account = None # type: ignore
try: # pragma: no cover - optional dependency that may not be installed in tests
import stripe # type: ignore
except ImportError: # pragma: no cover - gracefully handled when SDK missing
stripe = None # type: ignore
try: # pragma: no cover - optional dependency for AWS integrations
import boto3 # type: ignore
except ImportError: # pragma: no cover - gracefully handled when SDK missing
boto3 = None # type: ignore
MARKETING_ENUMERATION = os.path.join(
os.path.dirname(__file__), "..", "..", "daemon", "enumerations", "marketing_500.txt"
)
MARKETING_BENCHMARKS = os.path.join(
os.path.dirname(__file__), "..", "..", "daemon", "datasets", "marketing_benchmarks.json"
)
CJ_OFFERS_DATASET = os.path.join(
os.path.dirname(__file__), "..", "..", "daemon", "datasets", "cj_offers.json"
)
AMAZON_MARKETPLACE_DATASET = os.path.join(
os.path.dirname(__file__), "..", "..", "daemon", "datasets", "amazon_marketplace.json"
)
AWS_GROWTH_PLAYBOOKS = os.path.join(
os.path.dirname(__file__), "..", "..", "daemon", "datasets", "aws_growth_playbooks.json"
)
requests_spec = importlib.util.find_spec("requests")
requests = importlib.import_module("requests") if requests_spec else None
def _load_lines(path: str) -> List[str]:
with open(path, "r", encoding="utf-8") as handle:
return [line.strip() for line in handle if line.strip()]
@dataclass(slots=True)
class MarketingCampaign:
"""Structured representation for one marketing campaign idea."""
raw: str
channel: str
audience: str
offer: str
objective: str
@classmethod
def parse(cls, raw: str) -> "MarketingCampaign":
text = raw.strip().rstrip(".")
if not text.lower().startswith("campaign:"):
raise ValueError(f"Unsupported campaign format: {raw}")
body = text[len("Campaign:") :].strip()
try:
channel, remainder = body.split("targeting", 1)
audience, remainder = remainder.split("leveraging", 1)
offer, objective = remainder.split(" to ", 1)
except ValueError as exc: # pragma: no cover - enumeration should be well formed
raise ValueError(f"Malformed campaign sentence: {raw}") from exc
return cls(
raw=raw,
channel=channel.strip(),
audience=audience.strip(),
offer=offer.strip(),
objective=objective.strip().rstrip("."),
)
@dataclass(slots=True)
class CampaignMetrics:
"""Statistical expectations for a marketing campaign."""
expected_roi: float
expected_ctr: float
confidence: float
drivers: Dict[str, float]
class MarketingBenchmarks:
"""Loads benchmark statistics and produces campaign metrics."""
def __init__(self, baseline: Dict[str, float], components: Dict[str, Dict[str, Dict[str, float]]]):
self._baseline = baseline
self._components = components
@classmethod
def from_file(cls, path: str = MARKETING_BENCHMARKS) -> "MarketingBenchmarks":
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
baseline = payload.get("baseline", {"roi": 2.0, "ctr": 0.02, "confidence": 0.6})
components: Dict[str, Dict[str, Dict[str, float]]] = {}
for feature, entries in payload.items():
if feature == "baseline":
continue
components[feature] = {key.lower(): value for key, value in entries.items()}
return cls(baseline, components)
def _lookup(self, feature: str, value: str) -> Dict[str, float]:
cohort = self._components.get(feature, {})
return cohort.get(value.lower(), self._baseline)
def metrics_for(self, campaign: MarketingCampaign) -> CampaignMetrics:
weights = {
"channels": 0.32,
"audiences": 0.3,
"offers": 0.22,
"objectives": 0.16,
}
baseline_weight = max(0.0, 1.0 - sum(weights.values()))
base_roi = self._baseline.get("roi", 2.0)
base_ctr = self._baseline.get("ctr", 0.02)
base_conf = self._baseline.get("confidence", 0.6)
roi = base_roi * baseline_weight
ctr = base_ctr * baseline_weight
confidence = base_conf * baseline_weight
drivers: Dict[str, float] = {}
features = {
"channels": campaign.channel,
"audiences": campaign.audience,
"offers": campaign.offer,
"objectives": campaign.objective,
}
for feature, value in features.items():
stats = self._lookup(feature, value)
weight = weights[feature]
roi += stats.get("roi", base_roi) * weight
ctr += stats.get("ctr", base_ctr) * weight
confidence += stats.get("confidence", base_conf) * weight
drivers[feature.rstrip("s")] = stats.get("roi", base_roi)
return CampaignMetrics(
expected_roi=roi,
expected_ctr=ctr,
confidence=min(0.99, max(0.01, confidence)),
drivers=drivers,
)
@dataclass(slots=True)
class CampaignScore:
"""Composite score for a marketing campaign."""
campaign: MarketingCampaign
score: float
metrics: CampaignMetrics
keyword_coverage: Dict[str, int]
@dataclass(slots=True)
class PromptDiagnostics:
"""Structured analysis of a marketing prompt."""
stage: str
urgency: str
focus: str
recommended_kpis: List[str]
opportunities: List[str]
risk_signals: List[str]
keyword_summary: Dict[str, List[str]]
confidence: float
@dataclass(slots=True)
class GrowthExperiment:
"""Experiment plan produced by the Experiment Growth Grid (EGG) process."""
name: str
hypothesis: str
primary_metric: str
secondary_metrics: List[str]
playbook: List[str]
confidence: float
@dataclass(slots=True)
class 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 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]
@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
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 += 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
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))
demand_factor = max(75.0, avg_roi * 45.0 + avg_ctr * 1200)
projected_revenue = round(
listing.price_usd
* listing.commission_rate
* demand_factor
* (0.3 + intensity * 0.2)
/ 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,
projected_monthly_revenue=projected_revenue,
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
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():
return (
"BlockCypher height "
f"{data['height']} with {data['unconfirmed_count']} unconfirmed tx"
)
return None
def market_data(self, prompt: str, limit: int = 3) -> List[CryptoMarketSnapshot]:
assets = self._infer_assets(prompt, limit)
snapshots: List[CryptoMarketSnapshot] = []
blockstream_note = self._fetch_blockstream()
for asset in assets:
payload = self._fetch_coingecko(asset) or self._fetch_coincap(asset)
if payload is None:
fallback = self.FALLBACK_PRICES.get(asset) or self.FALLBACK_PRICES["bitcoin"]
payload = dict(fallback)
else:
payload = dict(payload)
source = payload.pop("source", "simulated")
notes = payload.pop("notes", None)
chain_note = self._fetch_blockcypher(asset)
combined_notes = " | ".join(
note for note in [notes, blockstream_note if asset == "bitcoin" else None, chain_note] if note
)
snapshots.append(
CryptoMarketSnapshot(
asset=asset,
price_usd=payload.get("price_usd"),
change_24h=payload.get("change_24h"),
market_cap=payload.get("market_cap"),
volume_24h=payload.get("volume_24h"),
source=source,
notes=combined_notes or None,
)
)
return snapshots
@dataclass(slots=True)
class AWSRevenueProgram:
"""AWS program metadata for revenue acceleration."""
program_name: str
focus: str
description: str
revenue_motion: str
primary_service: str
secondary_services: List[str]
automation_play: str
base_roi: float
focus_keywords: List[str]
@dataclass(slots=True)
class AWSRevenueRecommendation:
"""Recommendation payload produced by the AWS revenue process."""
program_name: str
focus: str
description: str
revenue_motion: str
primary_service: str
secondary_services: List[str]
automation_play: str
projected_roi: float
source: str
notes: str | None = None
class AWSGrowthPlaybookCatalog:
"""Loads AWS growth programs and matches them to prompts."""
def __init__(self, programs: Sequence[AWSRevenueProgram] | None = None):
if programs is not None:
self._programs = list(programs)
else:
with open(AWS_GROWTH_PLAYBOOKS, "r", encoding="utf-8") as handle:
data = json.load(handle)
self._programs = [AWSRevenueProgram(**entry) for entry in data]
def match_programs(
self, tokens: Sequence[str], limit: int
) -> List[Tuple[float, AWSRevenueProgram]]:
ranked: List[Tuple[float, AWSRevenueProgram]] = []
token_set = {token.lower() for token in tokens}
for program in self._programs:
haystack = (
f"{program.program_name} {program.description} {program.revenue_motion} "
f"{program.primary_service} {' '.join(program.secondary_services)}"
).lower()
score = 0.0
for token in token_set:
if token in haystack:
score += 1.2
if token in {kw.lower() for kw in program.focus_keywords}:
score += 1.6
if program.focus in token_set:
score += 0.5
ranked.append((score, program))
ranked.sort(key=lambda item: item[0], reverse=True)
return ranked[: max(1, limit)]
class AWSRevenueProcess:
"""Generates AWS-backed revenue motions and optionally validates via boto3."""
REQUIRED_ENV_VARS = {
"AWS_ACCESS_KEY_ID": "access_key_id",
"AWS_SECRET_ACCESS_KEY": "[credential removed]",
}
OPTIONAL_ENV_VARS = {
"AWS_SESSION_TOKEN": "session_token",
"AWS_REGION": "region",
"AWS_PROFILE": "profile",
}
def __init__(
self,
catalog: AWSGrowthPlaybookCatalog | None = None,
config: Dict[str, Any] | None = None,
) -> None:
self._catalog = catalog or AWSGrowthPlaybookCatalog()
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_id"),
aws_secret_access_key=[credential removed]"secret_access_key"),
aws_session_token=self._config.get("session_token"),
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 plays(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int,
) -> List[AWSRevenueRecommendation]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
matches = self._catalog.match_programs(tokens, limit)
if not matches:
return []
avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.5
avg_conf = mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.65
recommendations: List[AWSRevenueRecommendation] = []
for match_score, program in matches:
projected_roi = round(
program.base_roi + match_score * 0.2 + max(0.0, avg_roi - 2.0) * 0.15,
2,
)
projected_roi = max(projected_roi, program.base_roi)
rationale = (
f"Leverage {program.primary_service} with {', '.join(program.secondary_services)} "
f"to execute {program.revenue_motion.lower()}"
)
notes = None
if not self._session:
notes = (
"Offline AWS growth playbooks active; configure AWS credentials to validate deployment."
)
recommendations.append(
AWSRevenueRecommendation(
program_name=program.program_name,
focus=program.focus,
description=program.description,
revenue_motion=program.revenue_motion,
primary_service=program.primary_service,
secondary_services=program.secondary_services,
automation_play=program.automation_play,
projected_roi=max(projected_roi, avg_conf * program.base_roi),
source="aws_sdk" if self._session else "aws_playbooks",
notes=notes,
)
)
return recommendations
@dataclass(slots=True)
class GoogleCampaignPerformance:
"""Performance snapshot surfaced from the Google Ads API."""
campaign_name: str
clicks: int
impressions: int
conversions: float
cost_micros: int
ctr: float
cpc: float
roas: float
source: str
notes: str | None = None
@dataclass(slots=True)
class StripeMonetizationPlan:
"""Stripe checkout strategy synthesized from campaign intelligence."""
product_name: str
price_usd: float
billing_interval: str
currency: str
rationale: str
setup_steps: List[str]
source: str
notes: str | None = None
class StripeProcess:
"""Shapes campaign ideas into Stripe-ready monetization workflows."""
def __init__(self, api_key: str | None = None, account_id: str | None = None) -> None:
self._api_key = [credential removed] or os.getenv("STRIPE_API_KEY")
self._account_id = account_id or os.getenv("STRIPE_ACCOUNT_ID")
self._configured = bool(stripe is not None and self._api_key)
if self._configured and stripe is not None: # pragma: no cover - requires SDK
stripe.api_key = [credential removed]
@property
def configured(self) -> bool:
return self._configured
def _infer_billing_interval(self, prompt: str) -> str:
lowered = prompt.lower()
subscription_keywords = {
"subscription",
"recurring",
"membership",
"retainer",
"saas",
}
if any(keyword in lowered for keyword in subscription_keywords):
return "monthly"
if "annual" in lowered or "yearly" in lowered:
return "annual"
return "one_time"
def plans(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int = 3,
) -> List[StripeMonetizationPlan]:
if not campaigns:
raise ValueError("Stripe plans require at least one campaign score")
interval = self._infer_billing_interval(prompt)
plans: List[StripeMonetizationPlan] = []
for scored in campaigns[: max(1, limit)]:
base_price = max(29.0, scored.metrics.expected_roi * 45)
audience = scored.campaign.audience.lower()
if "enterprise" in audience:
base_price *= 1.6
elif "consumer" in audience:
base_price *= 0.75
if interval == "annual":
base_price *= 10.5
elif interval == "one_time":
base_price *= 1.25
price = round(base_price, 2)
rationale = (
f"Stripe-ready {interval.replace('_', ' ')} offer leveraging {scored.campaign.channel} "
f"for {scored.campaign.audience}."
)
if scored.metrics.confidence < 0.5:
rationale += " Calibrate pricing with phased rollout given confidence score."
setup_steps = [
"Create Product in Stripe aligned to the campaign offer.",
"Attach a Price with the recommended interval and USD currency.",
"Launch via Stripe Checkout or Payment Links to validate demand.",
]
if interval != "one_time":
setup_steps.append("Enable subscription trials and dunning automations.")
else:
setup_steps.append("Automate follow-up journeys via Stripe Webhooks or customer portal.")
source = "stripe_sdk" if self._configured else "simulated"
notes = None
if not self._configured:
notes = (
"Install the official stripe SDK and export STRIPE_API_KEY/STRIPE_ACCOUNT_ID "
"to activate live checkout orchestration."
)
plans.append(
StripeMonetizationPlan(
product_name=f"{scored.campaign.offer.title()} Checkout",
price_usd=price,
billing_interval=interval,
currency="usd",
rationale=rationale,
setup_steps=setup_steps,
source=source,
notes=notes,
)
)
return plans
class GoogleAdsProcess:
"""Bridges the Google Ads API to enrich ANI recommendations."""
REQUIRED_ENV_VARS = {
"GOOGLE_ADS_DEVELOPER_TOKEN": "developer_token",
"GOOGLE_ADS_CUSTOMER_ID": "customer_id",
}
OAUTH_ENV_VARS = {
"GOOGLE_ADS_CLIENT_ID": "client_id",
"GOOGLE_ADS_CLIENT_SECRET": "[credential removed]",
"GOOGLE_ADS_REFRESH_TOKEN": "refresh_token",
}
OPTIONAL_ENV_VARS = {
"GOOGLE_ADS_LOGIN_CUSTOMER_ID": "login_customer_id",
}
SERVICE_ACCOUNT_ENV_VARS = {
"GOOGLE_ADS_SERVICE_ACCOUNT_INFO": "service_account_info",
"GOOGLE_ADS_SERVICE_ACCOUNT_FILE": "service_account_file",
"GOOGLE_ADS_SERVICE_ACCOUNT_B64": "service_account_b64",
"GOOGLE_APPLICATION_CREDENTIALS": "service_account_file",
}
def __init__(self, config: Dict[str, Any] | None = None):
self._config = config or self._config_from_env()
self._client = self._build_client(self._config)
@classmethod
def _config_from_env(cls) -> Dict[str, Any]:
config: Dict[str, Any] = {}
for env_var, key in cls.REQUIRED_ENV_VARS.items():
value = os.getenv(env_var)
if value:
config[key] = value
for env_var, key in {**cls.OAUTH_ENV_VARS, **cls.OPTIONAL_ENV_VARS}.items():
value = os.getenv(env_var)
if value:
config[key] = value
for env_var, key in cls.SERVICE_ACCOUNT_ENV_VARS.items():
value = os.getenv(env_var)
if not value:
continue
if key == "service_account_b64":
try:
decoded = base64.b64decode(value).decode("utf-8")
config["service_account_info"] = json.loads(decoded)
except (ValueError, json.JSONDecodeError):
continue
elif key == "service_account_file":
try:
with open(value, "r", encoding="utf-8") as handle:
config["service_account_info"] = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError, OSError):
continue
elif key == "service_account_info":
try:
config["service_account_info"] = json.loads(value)
except json.JSONDecodeError:
continue
return config
@staticmethod
def _build_client(config: Dict[str, Any]) -> GoogleAdsClient | None:
if not config or "customer_id" not in config:
return None
if GoogleAdsClient is None:
return None
try: # pragma: no cover - relies on network credentials to execute
credentials = None
if config.get("service_account_info") and service_account is not None:
info = config["service_account_info"]
if isinstance(info, str):
try:
info = json.loads(info)
except json.JSONDecodeError:
info = None
if isinstance(info, dict):
scopes = ["https://www.googleapis.com/auth/adwords"]
credentials = service_account.Credentials.from_service_account_info(
info, scopes=scopes
)
client_kwargs = {
"developer_token": config.get("developer_token"),
"login_customer_id": config.get("login_customer_id"),
"use_proto_plus": True,
}
if credentials is not None:
return GoogleAdsClient.load_from_dict(client_kwargs, credentials=credentials)
# Fallback to OAuth credential set when present
if all(config.get(field) for field in ("client_id", "client_secret", "refresh_token")):
client_kwargs.update(
{
"client_id": config.get("client_id"),
"client_secret": config.get("client_secret"),
"refresh_token": config.get("refresh_token"),
}
)
return GoogleAdsClient.load_from_dict(client_kwargs)
except Exception: # pragma: no cover - fallback handled downstream
return None
@property
def configured(self) -> bool:
return bool(self._client)
def performance(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int,
) -> List[GoogleCampaignPerformance]:
limit = max(1, limit)
if self._client is None:
return self._simulate(prompt, campaigns, limit)
try: # pragma: no cover - executed when Google Ads credentials available
ga_service = self._client.get_service("GoogleAdsService")
query = self._build_query(limit)
response = ga_service.search(
customer_id=self._config.get("customer_id"),
query=query,
)
insights: List[GoogleCampaignPerformance] = []
for row in response:
metrics = row.metrics
campaign = row.campaign
clicks = int(getattr(metrics, "clicks", 0))
impressions = int(getattr(metrics, "impressions", 0))
conversions = float(getattr(metrics, "conversions", 0.0))
cost_micros = int(getattr(metrics, "cost_micros", 0))
ctr = float(getattr(metrics, "ctr", 0.0))
cpc = float(getattr(metrics, "average_cpc", 0.0))
roas = float(getattr(metrics, "conversions_value_per_cost", 0.0))
insights.append(
GoogleCampaignPerformance(
campaign_name=getattr(campaign, "name", "Campaign"),
clicks=clicks,
impressions=impressions,
conversions=conversions,
cost_micros=cost_micros,
ctr=ctr,
cpc=cpc,
roas=roas,
source="google_ads_api",
notes="Live Google Ads data",
)
)
if insights:
return insights
except Exception: # pragma: no cover - gracefully fallback on API errors
pass
return self._simulate(prompt, campaigns, limit)
@staticmethod
def _build_query(limit: int) -> str:
return (
"SELECT campaign.name, metrics.clicks, metrics.impressions, metrics.conversions, "
"metrics.cost_micros, metrics.ctr, metrics.average_cpc, metrics.conversions_value_per_cost "
"FROM campaign WHERE segments.date DURING LAST_30_DAYS "
"ORDER BY metrics.conversions DESC "
f"LIMIT {limit}"
)
def _simulate(
self, prompt: str, campaigns: Sequence[CampaignScore], limit: int
) -> List[GoogleCampaignPerformance]:
simulated: List[GoogleCampaignPerformance] = []
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
emphasis = 1 + len(set(tokens)) * 0.02
for index, scored in enumerate(campaigns[:limit]):
base_clicks = max(50, int(scored.score * 10))
clicks = int(base_clicks * emphasis)
impressions = max(clicks * 8, 500)
conversions = round(max(1.0, scored.metrics.expected_roi * 1.5), 2)
cost_micros = int(max(1_000_000, impressions * 120))
ctr = round(min(0.25, clicks / max(1, impressions)), 4)
cpc = round(cost_micros / max(1, clicks) / 1_000_000, 2)
roas = round(scored.metrics.expected_roi / max(0.1, cpc), 2)
simulated.append(
GoogleCampaignPerformance(
campaign_name=f"{scored.campaign.channel.title()} - Simulated",
clicks=clicks,
impressions=impressions,
conversions=conversions,
cost_micros=cost_micros,
ctr=ctr,
cpc=cpc,
roas=roas,
source="simulated",
notes=(
"Simulated from marketing benchmarks; set GOOGLE_ADS_* env vars "
"for live Google Ads enrichment."
),
)
)
if not simulated:
simulated.append(
GoogleCampaignPerformance(
campaign_name="Simulated Campaign",
clicks=120,
impressions=2400,
conversions=3.0,
cost_micros=2500000,
ctr=0.05,
cpc=0.5,
roas=4.2,
source="simulated",
notes="Fallback simulation due to missing campaigns",
)
)
return simulated
class CJAffiliateCatalog:
"""Loads CJ affiliate offers and surfaces relevant matches."""
def __init__(self, offers: Sequence[CJAffiliateOffer]):
self._offers = list(offers)
@classmethod
def from_file(cls, path: str = CJ_OFFERS_DATASET) -> "CJAffiliateCatalog":
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except FileNotFoundError:
return cls([])
except json.JSONDecodeError:
return cls([])
offers: List[CJAffiliateOffer] = []
for entry in payload if isinstance(payload, list) else []:
try:
offer = CJAffiliateOffer(
program_name=entry["program_name"],
advertiser_id=str(entry["advertiser_id"]),
category=entry.get("category", "Unknown"),
commission_rate=float(entry.get("commission_rate", 0.0)),
epc=float(entry.get("epc", 0.0)),
seven_day_epc=float(entry.get("seven_day_epc", entry.get("epc", 0.0))),
network_rank=int(entry.get("network_rank", 7)),
payout_terms=entry.get("payout_terms", "Net30"),
landing_url=entry.get("landing_url", ""),
description=entry.get("description", ""),
keywords=list(entry.get("keywords", [])),
)
except (KeyError, TypeError, ValueError):
continue
offers.append(offer)
return cls(offers)
def match_offers(
self,
tokens: Sequence[str],
limit: int = 3,
focus: str | None = None,
) -> List[Tuple[float, CJAffiliateOffer]]:
limit = max(1, limit)
scored: List[Tuple[float, CJAffiliateOffer]] = []
token_set = {token.lower() for token in tokens if token}
for offer in self._offers:
keywords = {keyword.lower() for keyword in offer.keywords}
description = offer.description.lower()
category = offer.category.lower()
score = offer.epc / 120.0
score += max(0.0, (6 - min(offer.network_rank, 6)) * 0.1)
for token in token_set:
if token in keywords:
score += 1.2
elif token in description or token in category:
score += 0.8
if focus and focus in category:
score += 0.5
scored.append((score, offer))
scored.sort(key=lambda item: item[0], reverse=True)
return scored[:limit]
class CJAffiliateProcess:
"""Transforms CJ affiliate data into monetization playbooks."""
REQUIRED_ENV_VARS = {
"CJ_DEVELOPER_KEY": "developer_key",
}
OPTIONAL_ENV_VARS = {
"CJ_WEBSITE_ID": "website_id",
"CJ_ACCOUNT_ID": "account_id",
}
def __init__(
self,
catalog: CJAffiliateCatalog | None = None,
config: Dict[str, Any] | None = None,
):
self._catalog = catalog or CJAffiliateCatalog.from_file()
self._config = config or self._config_from_env()
@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:
return "developer_key" in self._config
def offers(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int,
) -> List[CJAffiliateRecommendation]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
focus = None
if campaigns:
focus = campaigns[0].campaign.channel.lower().split()[0]
matches = self._catalog.match_offers(tokens, limit, focus)
if not matches:
return []
avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.5
avg_ctr = mean([score.metrics.expected_ctr for score in campaigns]) if campaigns else 0.03
enriched: List[CJAffiliateRecommendation] = []
for match_score, offer in matches:
keyword_matches = sorted({token for token in tokens if token in {k.lower() for k in offer.keywords}})
intensity = max(0.15, min(1.8, match_score / 4.0))
projected_epc = round(offer.epc * (1 + intensity * 0.18 + avg_ctr * 4), 2)
projected_revenue = round(projected_epc * max(5.0, avg_roi * 6), 2)
if keyword_matches:
rationale = (
f"Matches {', '.join(keyword_matches)} demand with {offer.category.lower()} payouts."
)
else:
rationale = f"High network rank offer aligned to {offer.category.lower()} intent."
source = "cj_network" if self.configured else "cj_catalog"
notes = None
if not self.configured:
notes = (
"Simulated from CJ offer corpus; set CJ_DEVELOPER_KEY and related env vars "
"to activate live API harvesting."
)
enriched.append(
CJAffiliateRecommendation(
program_name=offer.program_name,
advertiser_id=offer.advertiser_id,
category=offer.category,
commission_rate=round(offer.commission_rate, 4),
epc=round(offer.epc, 2),
seven_day_epc=round(offer.seven_day_epc, 2),
projected_epc=projected_epc,
projected_monthly_revenue=projected_revenue,
network_rank=offer.network_rank,
payout_terms=offer.payout_terms,
landing_url=offer.landing_url,
rationale=rationale,
source=source,
notes=notes,
)
)
return enriched
class MarketingIntelligenceEngine:
"""Ranks marketing campaigns against free-form prompts."""
def __init__(
self,
campaigns: Sequence[MarketingCampaign],
benchmarks: MarketingBenchmarks | None = None,
google_process: GoogleAdsProcess | None = None,
cj_process: CJAffiliateProcess | None = None,
crypto_process: CryptoMarketProcess | None = None,
stripe_process: StripeProcess | None = None,
amazon_process: AmazonMarketplaceProcess | None = None,
aws_process: AWSRevenueProcess | None = None,
):
if not campaigns:
raise ValueError("MarketingIntelligenceEngine requires at least one campaign")
self._campaigns = list(campaigns)
self._benchmarks = benchmarks or MarketingBenchmarks.from_file()
self._google_process = google_process or GoogleAdsProcess()
self._cj_process = cj_process or CJAffiliateProcess()
self._crypto_process = crypto_process or CryptoMarketProcess()
self._stripe_process = stripe_process or StripeProcess()
self._amazon_process = amazon_process or AmazonMarketplaceProcess()
self._aws_process = aws_process or AWSRevenueProcess()
@classmethod
def from_file(
cls,
path: str = MARKETING_ENUMERATION,
benchmark_path: str = MARKETING_BENCHMARKS,
) -> "MarketingIntelligenceEngine":
lines = _load_lines(path)
campaigns = [MarketingCampaign.parse(line) for line in lines]
benchmarks = MarketingBenchmarks.from_file(benchmark_path)
return cls(campaigns, benchmarks)
@staticmethod
def _keyword_set(prompt: str) -> List[str]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
filtered = [token for token in tokens if len(token) > 2]
return filtered or ["campaign"]
def recommend(self, prompt: str, limit: int = 3) -> List[MarketingCampaign]:
tokens = self._keyword_set(prompt)
def score(campaign: MarketingCampaign) -> tuple[int, int, str]:
haystacks = {
"channel": campaign.channel.lower(),
"audience": campaign.audience.lower(),
"offer": campaign.offer.lower(),
"objective": campaign.objective.lower(),
}
weighted = 0
matched_features: set[str] = set()
for token in tokens:
for feature, text in haystacks.items():
if token in text:
matched_features.add(feature)
weight = 4 if feature == "audience" else 3 if feature == "offer" else 2
weighted += weight
# fallback to slight preference for broader coverage even if nothing matched
coverage_bonus = len(matched_features)
return (-(weighted + coverage_bonus), len(campaign.raw), campaign.raw)
ranked = sorted(self._campaigns, key=score)
return ranked[: max(1, limit)]
def recommend_statistical(self, prompt: str, limit: int = 3) -> List[CampaignScore]:
tokens = self._keyword_set(prompt)
scored: List[CampaignScore] = []
for campaign in self._campaigns:
metrics = self._benchmarks.metrics_for(campaign)
keyword_hits: Dict[str, int] = {"channel": 0, "audience": 0, "offer": 0, "objective": 0}
haystacks = {
"channel": campaign.channel.lower(),
"audience": campaign.audience.lower(),
"offer": campaign.offer.lower(),
"objective": campaign.objective.lower(),
}
match_bonus = 0.0
for token in tokens:
for feature, text in haystacks.items():
if token in text:
keyword_hits[feature] += 1
match_bonus += 0.35
roi_component = metrics.expected_roi * 0.6
ctr_component = metrics.expected_ctr * 80.0 * 0.25
confidence_component = metrics.confidence * 10.0 * 0.1
keyword_component = match_bonus
composite = roi_component + ctr_component + confidence_component + keyword_component
scored.append(
CampaignScore(
campaign=campaign,
score=composite,
metrics=metrics,
keyword_coverage=keyword_hits,
)
)
ranked = sorted(scored, key=lambda entry: entry.score, reverse=True)
return ranked[: max(1, limit)]
def summarize(self, prompt: str, campaigns: Sequence[MarketingCampaign]) -> str:
tokens = self._keyword_set(prompt)
highlights = ", ".join(sorted(set(tokens)))
count = len(campaigns)
return f"Surface {count} campaign plays tuned for {highlights}."
@staticmethod
def _infer_stage(tokens: Sequence[str]) -> Tuple[str, List[str]]:
mapping = {
"awareness": {"awareness", "reach", "impressions", "brand", "launch", "attention"},
"consideration": {"consideration", "education", "webinar", "nurture", "pipeline", "engagement"},
"conversion": {
"conversion",
"roi",
"ctr",
"leads",
"demos",
"trial",
"customers",
"acquisition",
},
"retention": {"retention", "renewal", "expansion", "upsell", "loyalty", "churn"},
}
hits: Dict[str, List[str]] = {key: [] for key in mapping}
for token in tokens:
for stage, keywords in mapping.items():
if token in keywords:
hits[stage].append(token)
stage = max(mapping, key=lambda key: len(hits[key]))
if all(len(values) == 0 for values in hits.values()):
stage = "conversion"
return stage, hits[stage]
@staticmethod
def _infer_urgency(prompt: str) -> str:
lowered = prompt.lower()
if any(keyword in lowered for keyword in ["urgent", "asap", "immediately", "right away", "now"]):
return "urgent"
if any(keyword in lowered for keyword in ["this quarter", "q4", "deadline", "accelerate"]):
return "time-sensitive"
return "standard"
@staticmethod
def _infer_focus(tokens: Sequence[str]) -> Tuple[str, List[str]]:
focus_buckets = {
"enterprise b2b": {"enterprise", "cxo", "c-suite", "fortune", "b2b", "saas"},
"startup founders": {"founder", "startup", "scaleup", "seed", "series"},
"consumer": {"consumer", "b2c", "retail", "shopper", "app"},
"developer": {"developer", "devops", "engineer", "technical", "api"},
}
hits: Dict[str, List[str]] = {bucket: [] for bucket in focus_buckets}
for token in tokens:
for focus, keywords in focus_buckets.items():
if token in keywords:
hits[focus].append(token)
focus = max(focus_buckets, key=lambda key: len(hits[key]))
if len(hits[focus]) == 0:
return "broad b2b", []
return focus, hits[focus]
@staticmethod
def _recommend_kpis(stage: str) -> List[str]:
mapping = {
"awareness": ["impressions", "share of voice", "reach growth"],
"consideration": ["content engagement", "marketing qualified leads", "pipeline velocity"],
"conversion": ["sales qualified leads", "demo-to-close rate", "customer acquisition cost"],
"retention": ["net revenue retention", "renewal rate", "expansion ARR"],
}