A continuation of Janady13/AI_Brain / services/seeder_api/intelligence.py. The complete source identity, every recorded occurrence, and linked equations are on its first page.
Source transcript · continued
Download complete source text ↓4,340 lines · 162.2 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
score += 0.35
if listing.shipping_days <= 3:
score += 0.25
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 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",
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: EbayMarketplaceCatalog | None = None,
config: Dict[str, Any] | None = None,
) -> None:
self._catalog = catalog or EbayMarketplaceCatalog()
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 requests is None:
return None
try:
session = requests.Session()
session.headers.update({"User-Agent": "ai-brain-marketing-ani/1.0"})
return session
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[EbayRevenuePlay]:
) -> 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.0
avg_confidence = (
mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.5
)
recommendations: List[EbayRevenuePlay] = []
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.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,
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
class BudgetAllocatorProcess:
"""Turns campaign rankings into channel-level budget plans."""
def __init__(
self,
profiles: Sequence[BudgetProfile] | None = None,
dataset_path: str = CHANNEL_INVESTMENTS_DATASET,
default_budget: float = 25000.0,
):
self._profiles = list(profiles) if profiles is not None else self._profiles_from_file(dataset_path)
self._default_budget = default_budget
@staticmethod
def _profiles_from_file(path: str) -> List[BudgetProfile]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except FileNotFoundError:
payload = {"profiles": []}
profiles: List[BudgetProfile] = []
for entry in payload.get("profiles", []):
profiles.append(
BudgetProfile(
channel=entry.get("channel", "Integrated Marketing"),
keywords=[keyword.lower() for keyword in entry.get("keywords", [])],
baseline_cpc=float(entry.get("baseline_cpc", 4.0)),
baseline_cpl=float(entry.get("baseline_cpl", 150.0)),
avg_conversion_rate=float(entry.get("avg_conversion_rate", 0.045)),
avg_roi=float(entry.get("avg_roi", 3.2)),
recommended_monthly_spend=float(entry.get("recommended_monthly_spend", 20000.0)),
)
)
if not profiles:
profiles.append(
BudgetProfile(
channel="Integrated Marketing",
keywords=["integrated", "multi-channel"],
baseline_cpc=4.0,
baseline_cpl=150.0,
avg_conversion_rate=0.045,
avg_roi=3.0,
recommended_monthly_spend=20000.0,
)
)
return profiles
def _match_profile(self, channel: str) -> BudgetProfile:
lowered = channel.lower()
best: BudgetProfile | None = None
best_score = -1
for profile in self._profiles:
score = 0
if profile.channel.lower() in lowered:
score += 3
for keyword in profile.keywords:
if keyword in lowered:
score += 1
if score > best_score:
best = profile
best_score = score
if best is None:
best = BudgetProfile(
channel=channel,
keywords=[channel.lower()],
baseline_cpc=4.5,
baseline_cpl=165.0,
avg_conversion_rate=0.042,
avg_roi=3.1,
recommended_monthly_spend=self._default_budget,
)
return best
@staticmethod
def _parse_budget_from_prompt(prompt: str) -> float | None:
pattern = re.compile(r"\$?([0-9][0-9,]*)(?:\s*(k|m|b|million|billion|thousand))?", re.IGNORECASE)
matches = pattern.findall(prompt)
if not matches:
return None
inferred: List[float] = []
for value_str, multiplier in matches:
value = float(value_str.replace(",", ""))
if multiplier:
token = multiplier.lower()
if token in {"k", "thousand"}:
value *= 1_000
elif token in {"m", "million"}:
value *= 1_000_000
elif token in {"b", "billion"}:
value *= 1_000_000_000
elif value < 100:
value *= 1_000
inferred.append(value)
return max(inferred) if inferred else None
def _infer_budget(self, prompt: str, profiles: Sequence[BudgetProfile]) -> float:
parsed = self._parse_budget_from_prompt(prompt)
if parsed is not None and parsed > 0:
return float(parsed)
if profiles:
average_spend = mean(profile.recommended_monthly_spend for profile in profiles)
return float(max(average_spend, self._default_budget))
return self._default_budget
@staticmethod
def _allocation_weight(score: CampaignScore, profile: BudgetProfile) -> float:
roi_component = score.metrics.expected_roi * 0.45
ctr_component = score.metrics.expected_ctr * 100.0 * 0.2
profile_component = profile.avg_roi * 0.25
efficiency_component = 240.0 / max(profile.baseline_cpl, 60.0)
return max(0.1, roi_component + ctr_component + profile_component + efficiency_component)
@staticmethod
def _summarize_plan(budget: float, expected_revenue: float, count: int) -> str:
if budget <= 0 or expected_revenue <= 0:
return "No budgetable campaigns identified."
roi = expected_revenue / budget if budget else 0
return (
f"Deploy ${budget:,.0f} across {count} channels targeting ~${expected_revenue:,.0f} revenue "
f"(~{roi:.1f}x ROI)."
)
@staticmethod
def _payback_weeks(budget: float, expected_revenue: float) -> float:
profit = expected_revenue - budget
if budget <= 0:
return 0.0
if profit <= 0:
return 16.0
weeks = (budget / profit) * 4
return round(max(2.0, min(26.0, weeks)), 1)
def plan(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int = 3,
) -> BudgetPlan:
if not campaigns:
raise ValueError("plan requires at least one campaign score")
selected = list(campaigns[: max(1, limit)])
profiles = [self._match_profile(score.campaign.channel) for score in selected]
total_budget = self._infer_budget(prompt, profiles)
weighted: List[Tuple[CampaignScore, BudgetProfile, float]] = []
for score, profile in zip(selected, profiles):
weight = self._allocation_weight(score, profile)
weighted.append((score, profile, weight))
total_weight = sum(entry[2] for entry in weighted)
allocations: List[BudgetAllocation] = []
running_total = 0.0
for index, (score, profile, weight) in enumerate(weighted):
share = weight / total_weight if total_weight else 1.0 / len(weighted)
allocation = round(total_budget * share, 2)
if index == len(weighted) - 1:
allocation = round(total_budget - running_total, 2)
running_total += allocation
expected_leads = round(allocation / max(profile.baseline_cpl, 1.0), 1)
blended_roi = max(1.05, (score.metrics.expected_roi + profile.avg_roi) / 2)
expected_revenue = round(allocation * blended_roi, 2)
rationale = (
f"Amplify {score.campaign.channel} for {score.campaign.audience} to accelerate "
f"{score.campaign.objective}."
)
notes = None
if profile.channel.lower() not in score.campaign.channel.lower():
notes = (
f"Mapped to {profile.channel} benchmarks using keyword similarity; override dataset "
"to tune allocation."
)
allocations.append(
BudgetAllocation(
channel=score.campaign.channel,
audience_focus=score.campaign.audience,
offer=score.campaign.offer,
objective=score.campaign.objective,
allocation_usd=allocation,
allocation_weight=round(share, 3),
expected_leads=expected_leads,
expected_revenue=expected_revenue,
rationale=rationale,
source="budget_optimizer",
notes=notes,
)
)
total_expected_revenue = round(sum(entry.expected_revenue for entry in allocations), 2)
expected_roi = round(total_expected_revenue / total_budget, 2) if total_budget else 0.0
payback_weeks = self._payback_weeks(total_budget, total_expected_revenue)
summary = self._summarize_plan(total_budget, total_expected_revenue, len(allocations))
return BudgetPlan(
total_budget=round(total_budget, 2),
allocations=allocations,
expected_revenue=total_expected_revenue,
expected_roi=expected_roi,
payback_period_weeks=payback_weeks,
summary=summary,
)
@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
@dataclass(slots=True)
class PayPalMonetizationPlan:
"""PayPal commerce strategy to operationalize campaign monetization."""
product_name: str
price_usd: float
checkout_flow: str
payout_schedule: str
rationale: str
setup_steps: List[str]
source: str
notes: str | None = None
class PayPalProcess:
"""Designs PayPal-ready offers so campaigns translate into paychecks."""
def __init__(
self,
client_id: str | None = None,
client_secret: str | None = None,
merchant_id: str | None = None,
environment: str | None = None,
) -> None:
self._client_id = client_id or os.getenv("PAYPAL_CLIENT_ID")
self._client_secret = [credential removed] or os.getenv("PAYPAL_CLIENT_SECRET")
self._merchant_id = merchant_id or os.getenv("PAYPAL_MERCHANT_ID")
self._environment = (environment or os.getenv("PAYPAL_ENVIRONMENT", "sandbox")).lower()
self._configured = bool(
PayPalHttpClient is not None and self._client_id and self._client_secret
)
self._client: PayPalHttpClient | None = None
if self._configured and PayPalHttpClient is not None: # pragma: no cover - requires SDK
if self._environment in {"live", "production"} and LiveEnvironment is not None:
env = LiveEnvironment(self._client_id, self._client_secret)
else:
env = SandboxEnvironment(self._client_id, self._client_secret)
self._client = PayPalHttpClient(env)
@property
def configured(self) -> bool:
return self._configured
def _infer_checkout_flow(self, prompt: str) -> str:
lowered = prompt.lower()
if any(keyword in lowered for keyword in ["subscription", "membership", "recurring"]):
return "subscription"
if any(keyword in lowered for keyword in ["invoice", "retainer", "contract"]):
return "invoice"
return "one_time"
def _infer_payout_schedule(self, prompt: str) -> str:
lowered = prompt.lower()
if any(keyword in lowered for keyword in ["daily", "paycheck", "cash flow"]):
return "daily"
if any(keyword in lowered for keyword in ["weekly", "bi-weekly", "payout"]):
return "weekly"
return "rolling"
def plans(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int = 3,
) -> List[PayPalMonetizationPlan]:
if not campaigns:
raise ValueError("PayPal plans require at least one campaign score")
flow = self._infer_checkout_flow(prompt)
payout_schedule = self._infer_payout_schedule(prompt)
plans: List[PayPalMonetizationPlan] = []
for scored in campaigns[: max(1, limit)]:
base_price = max(19.0, scored.metrics.expected_roi * 35)
if "enterprise" in scored.campaign.audience.lower():
base_price *= 1.4
elif "consumer" in scored.campaign.audience.lower():
base_price *= 0.8
if flow == "subscription":
base_price *= 9.5
elif flow == "invoice":
base_price *= 4.5
price = round(base_price, 2)
rationale = (
f"PayPal {flow.replace('_', ' ')} monetization aligned to {scored.campaign.channel} "
f"for {scored.campaign.audience}."
)
if payout_schedule == "daily":
rationale += " Configured for accelerated daily paychecks."
setup_steps = [
"Create Product or Plan in PayPal matching the campaign offer.",
f"Configure {flow} checkout with contextual upsells.",
f"Enable payouts on a {payout_schedule} cadence for predictable cashflow.",
]
if self._merchant_id:
setup_steps.append(
f"Link merchant account {self._merchant_id} to PayPal Payouts for revenue capture."
)
else:
setup_steps.append("Map payouts to finance workflows for accounting reconciliation.")
source = "paypal_sdk" if self._configured else "simulated"
notes = None
if not self._configured:
notes = (
"Install paypalcheckoutsdk and export PAYPAL_CLIENT_ID/PAYPAL_CLIENT_SECRET "
"to activate live PayPal revenue collection."
)
plans.append(
PayPalMonetizationPlan(
product_name=f"{scored.campaign.offer.title()} PayPal Flow",
price_usd=price,
checkout_flow=flow,
payout_schedule=payout_schedule,
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 InfluencerCollaborationProcess:
"""Matches prompts with influencer partners to accelerate revenue loops."""
OPTIONAL_ENV_VARS = {
"INFLUENCER_NETWORK_API_KEY": "[credential removed]",
"INFLUENCER_NETWORK_WORKSPACE": "workspace",
}
def __init__(
self,
catalog: InfluencerNetworkCatalog | None = None,
config: Dict[str, Any] | None = None,
) -> None:
self._catalog = catalog or InfluencerNetworkCatalog()
self._config = config or self._config_from_env()
@classmethod
def _config_from_env(cls) -> Dict[str, Any]:
config: Dict[str, Any] = {}