A continuation of Janady13/AI_Brain / services/seeder_api/intelligence.py. The complete source identity, every recorded occurrence, and linked equations are on its first page.
Source transcript · continued
Download complete source text ↓5,617 lines · 212.6 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
@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,
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] = {}
for env_var, key in cls.OPTIONAL_ENV_VARS.items():
value = os.getenv(env_var)
if value:
config[key] = value
return config
@property
def configured(self) -> bool:
return "api_key" in self._config
def recommendations(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int,
) -> List[InfluencerRecommendation]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
focus = None
if campaigns:
primary = campaigns[0].campaign
focus = f"{primary.channel} {primary.audience}"
matches = self._catalog.match_partners(tokens, focus, limit)
if not matches:
return []
avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.2
avg_ctr = mean([score.metrics.expected_ctr for score in campaigns]) if campaigns else 0.03
avg_confidence = (
mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.55
)
reach_multiplier = max(1.1, 1 + (avg_confidence - 0.5))
recs: List[InfluencerRecommendation] = []
for index, (match_score, partner) in enumerate(matches):
collab_formats = partner.collab_formats or ["sponsored integration"]
collab_format = collab_formats[index % len(collab_formats)]
estimated_reach = int(
round(partner.audience_size * partner.engagement_rate * reach_multiplier)
)
estimated_reach = max(estimated_reach, int(partner.audience_size * 0.02))
lead_rate = max(0.015, avg_ctr * 1.15 + partner.engagement_rate / 3)
projected_leads = max(12, int(round(estimated_reach * lead_rate)))
investment = partner.avg_cpm * estimated_reach / 1000.0
revenue_per_lead = max(95.0, avg_roi * 120.0)
projected_revenue = round(projected_leads * revenue_per_lead, 2)
projected_roi = (
round(projected_revenue / investment, 2) if investment else round(avg_roi * 3, 2)
)
keyword_matches = sorted(
{
token
for token in tokens
if token in {keyword.lower() for keyword in partner.keywords}
}
)
if keyword_matches:
rationale = (
"Maps "
+ "/".join(keyword_matches[:3])
+ f" intent into {partner.platform} {collab_format} with {partner.niche.lower()} authority."
)
else:
rationale = (
f"Pairs {partner.niche.lower()} reach with {partner.conversion_focus.lower()} focus "
"to amplify campaign demand."
)
if partner.case_study:
rationale += f" Proof: {partner.case_study}"
source = "influencer_api" if self.configured else "influencer_catalog"
notes = None
if not self.configured:
notes = (
"Simulated from influencer partner corpus; set INFLUENCER_NETWORK_API_KEY to "
"activate live network sourcing."
)
recs.append(
InfluencerRecommendation(
name=partner.name,
platform=partner.platform,
niche=partner.niche,
collab_format=collab_format,
audience_size=partner.audience_size,
engagement_rate=round(partner.engagement_rate, 4),
estimated_reach=estimated_reach,
projected_cpm=round(partner.avg_cpm, 2),
projected_leads=projected_leads,
projected_revenue=projected_revenue,
projected_roi=projected_roi,
rationale=rationale,
source=source,
notes=notes,
)
)
return recs
class AutomationSprintPlanner:
"""Builds a three-hour automation sprint tailored to the prompt context."""
def __init__(
self,
templates: Dict[str, Any] | None = None,
defaults: Dict[str, Any] | None = None,
):
data = templates or self._load_templates()
self._templates = data.get("stages", {})
self._overlays = data.get("focus_overlays", {})
base_defaults = data.get("defaults", {}).copy()
if defaults:
base_defaults.update(defaults)
self._defaults = base_defaults
@staticmethod
def _load_templates(path: str = AUTOMATION_SPRINTS_DATASET) -> Dict[str, Any]:
with open(path, "r", encoding="utf-8") as handle:
return json.load(handle)
def _stage_templates(self, stage: str) -> List[Dict[str, Any]]:
if stage in self._templates:
return list(self._templates[stage])
if "conversion" in self._templates:
return list(self._templates["conversion"])
for entries in self._templates.values():
if entries:
return list(entries)
return []
def _focus_overlays(self, focus: str) -> List[Dict[str, Any]]:
lowered = focus.lower()
for key, tasks in self._overlays.items():
if key in lowered:
return list(tasks)
return []
@staticmethod
def _safe_format(template: str, context: Dict[str, str]) -> str:
class _SafeDict(dict):
def __missing__(self, key: str) -> str: # pragma: no cover - defensive fallback
return "{" + key + "}"
return template.format_map(_SafeDict(context))
@staticmethod
def _campaign_context(
campaigns: Sequence[CampaignScore], index: int
) -> Dict[str, str]:
if campaigns:
primary = campaigns[0].campaign
reference = campaigns[index % len(campaigns)].campaign
secondary = campaigns[1 % len(campaigns)].campaign if len(campaigns) > 1 else primary
return {
"channel": reference.channel,
"offer": reference.offer,
"audience": reference.audience,
"objective": reference.objective,
"primary_channel": primary.channel,
"primary_offer": primary.offer,
"primary_audience": primary.audience,
"primary_objective": primary.objective,
"support_channel": secondary.channel,
"support_offer": secondary.offer,
}
return {
"channel": "omnichannel automation",
"offer": "flagship offer",
"audience": "core buyers",
"objective": "pipeline velocity",
"primary_channel": "omnichannel",
"primary_offer": "flagship offer",
"primary_audience": "core buyers",
"primary_objective": "pipeline velocity",
"support_channel": "email",
"support_offer": "activation play",
}
def plan(
self,
prompt: str,
diagnostics: PromptDiagnostics,
campaigns: Sequence[CampaignScore],
limit: int,
) -> AutomationSprint:
block = int(self._defaults.get("block_minutes", 30))
total_minutes = int(self._defaults.get("total_minutes", 180))
blocks = max(1, total_minutes // max(1, block))
templates = self._stage_templates(diagnostics.stage)
overlays = self._focus_overlays(diagnostics.focus)
if overlays:
templates.extend(overlays)
if not templates:
templates = [
{
"activity": "Audit acquisition funnel touchpoints",
"outcome": "Document automation opportunities",
"tooling": ["Notion"],
"kpis": ["pipeline velocity"],
"automation_level": "assist",
}
]
tasks: List[AutomationTask] = []
summary_drivers: List[str] = []
for idx in range(blocks):
template = templates[idx % len(templates)]
context = self._campaign_context(campaigns, idx)
context["stage"] = diagnostics.stage
context["focus"] = diagnostics.focus
activity = self._safe_format(template.get("activity", "Execute automation sprint"), context)
outcome = self._safe_format(template.get("outcome", "Drive measurable impact"), context)
tooling = list(template.get("tooling", []))
kpis = [kpi.lower() for kpi in template.get("kpis", [])]
automation_level = template.get("automation_level", "assist")
start_minute = idx * block
end_minute = start_minute + block
source = self._defaults.get("source", "automation_playbooks")
tasks.append(
AutomationTask(
start_minute=start_minute,
end_minute=end_minute,
activity=activity,
outcome=outcome,
tooling=tooling,
kpis=kpis,
automation_level=automation_level,
source=source,
)
)
channel = context.get("channel")
if channel and channel not in summary_drivers:
summary_drivers.append(channel)
channels_summary = ", ".join(summary_drivers[:3]) or "omnichannel"
summary = (
f"Three-hour automation sprint targeting {diagnostics.stage} intent for {diagnostics.focus} focus "
f"anchored in {channels_summary} plays."
)
notes = self._defaults.get("notes")
return AutomationSprint(
total_minutes=blocks * block,
stage=diagnostics.stage,
focus=diagnostics.focus,
summary=summary,
tasks=tasks,
notes=notes,
source=self._defaults.get("source", "automation_playbooks"),
)
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,
stock_process: StockMarketProcess | None = None,
stripe_process: StripeProcess | None = None,
paypal_process: PayPalProcess | None = None,
amazon_process: AmazonMarketplaceProcess | None = None,
ebay_process: EbayMarketplaceProcess | None = None,
aws_process: AWSRevenueProcess | None = None,
budget_process: BudgetAllocatorProcess | None = None,
automation_process: AutomationSprintPlanner | None = None,
learning_process: LearningLoopProcess | None = None,
competitive_process: CompetitiveIntelligenceProcess | None = None,
influencer_process: InfluencerCollaborationProcess | None = None,
scenario_process: "ScenarioPlanningProcess" | None = None,
statistical_process: StatisticalReasoningProcess | None = None,
cognitive_process: CognitiveArchitectureProcess | None = None,
agentic_process: AgenticSystemsProcess | 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._stock_process = stock_process or StockMarketProcess()
self._stripe_process = stripe_process or StripeProcess()
self._paypal_process = paypal_process or PayPalProcess()
self._amazon_process = amazon_process or AmazonMarketplaceProcess()
self._ebay_process = ebay_process or EbayMarketplaceProcess()
self._aws_process = aws_process or AWSRevenueProcess()
self._budget_process = budget_process or BudgetAllocatorProcess()
self._automation_process = automation_process or AutomationSprintPlanner()
self._learning_process = learning_process or LearningLoopProcess()
self._competitive_process = competitive_process or CompetitiveIntelligenceProcess()
self._influencer_process = influencer_process or InfluencerCollaborationProcess()
self._scenario_process = scenario_process or ScenarioPlanningProcess()
self._statistical_process = statistical_process or StatisticalReasoningProcess()
self._cognitive_process = cognitive_process or CognitiveArchitectureProcess()
self._agentic_process = agentic_process or AgenticSystemsProcess()
@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"],
}
return mapping.get(stage, ["pipeline velocity", "conversion rate"])
@staticmethod
def _recommend_opportunities(stage: str, focus: str) -> List[str]:
opportunities = {
"awareness": [
"Anchor thought-leadership content to prime the market",