A continuation of Janady13/AI_Brain / services/seeder_api/intelligence.py. The complete source identity, every recorded occurrence, and linked equations are on its first page.
Source transcript · continued
Download complete source text ↓9,479 lines · 364.7 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
try:
scenarios.append(
QuantumScenarioTemplate(
name=raw["name"],
domain=raw["domain"],
objective=raw["objective"],
gate_sequence=list(raw.get("gate_sequence", [])),
target_probability=float(raw.get("target_probability", 0.5)),
stressors=list(raw.get("stressors", [])),
metrics=list(raw.get("metrics", [])),
integration_points=list(raw.get("integration_points", [])),
playbook=list(raw.get("playbook", [])),
notes=raw.get("notes"),
)
)
except (KeyError, TypeError, ValueError):
continue
if not scenarios:
continue
capabilities.append(
QuantumSimulatorCapability(
simulator=entry.get("simulator", "Quantum Simulator"),
provider=entry.get("provider", "open_source"),
description=entry.get(
"description",
"Quantum-inspired simulator for revenue stress testing.",
),
training_focus=list(entry.get("training_focus", [])),
evaluation_metrics=list(entry.get("evaluation_metrics", [])),
orchestration=list(entry.get("orchestration", [])),
dependencies=list(entry.get("dependencies", [])),
docs=list(entry.get("docs", [])),
scenarios=scenarios,
source=entry.get("source", "quantum_simulation_catalog"),
notes=entry.get("notes"),
)
)
if not capabilities:
fallback_scenarios: List[QuantumScenarioTemplate] = []
for domain in sorted(cls.REQUIRED_DOMAINS):
fallback_scenarios.append(
QuantumScenarioTemplate(
name=f"{domain.title()} Stress Harness",
domain=domain,
objective=f"Baseline quantum stress for {domain.replace('_', ' ')} stacks.",
gate_sequence=["H", "RX(pi/4)", "RZ(pi/6)", "H"],
target_probability=0.6,
stressors=["throughput", "latency"],
metrics=["stability_index"],
integration_points=[],
playbook=["Run fallback simulator to validate coverage."],
notes="Fallback coverage generated when dataset missing.",
)
)
capabilities.append(
QuantumSimulatorCapability(
simulator="Fallback Quantum Simulator",
provider="deterministic",
description="Deterministic single-qubit simulator for baseline stress coverage.",
training_focus=["Quantum-inspired stress heuristics"],
evaluation_metrics=["stability_index"],
orchestration=["Upgrade dataset to unlock provider-specific playbooks."],
dependencies=[],
docs=[],
scenarios=fallback_scenarios,
source="quantum_simulation_fallback",
notes="Generated offline fallback to preserve ANI readiness.",
)
)
return capabilities
def plans(
self,
prompt: str,
campaigns: Sequence["CampaignScore"],
limit: int = 1,
) -> List[QuantumStressTestPlan]:
if not campaigns:
raise ValueError("Quantum simulator plans require campaign scores")
prompt_tokens = self._tokenize(prompt)
campaign_tokens = self._tokenize(
" ".join(
" ".join(
[
score.campaign.channel,
score.campaign.audience,
score.campaign.offer,
score.campaign.objective,
]
)
for score in campaigns
)
)
plans: List[Tuple[float, QuantumStressTestPlan]] = []
for capability in self._capabilities:
stress_tests: List[QuantumStressTest] = []
coverage: set[str] = set()
stability_scores: List[float] = []
per_test_agents: List[int] = []
per_test_nodes: List[int] = []
per_test_compression: List[float] = []
per_test_shots: List[int] = []
domain_hits = 0
domain_stability_map: Dict[str, float] = {}
for template in capability.scenarios:
coverage.add(template.domain.lower())
measurement = self._simulator.run(template.gate_sequence)
stability = 1.0 - abs(measurement - template.target_probability)
stability = max(0.0, min(1.0, stability))
stability_scores.append(stability)
domain_stability_map[template.domain.lower()] = round(stability, 4)
shots_used = getattr(
self._simulator,
"last_shots",
getattr(self._simulator, "shots", 1024),
)
stability_floor = max(0.05, stability)
agents_per_node = max(1, int(round(1_000_000 * stability_floor)))
nodes_for_one_million = max(1, math.ceil(1_000_000 / agents_per_node))
compression_ratio = round(
min(1.0, stability_floor / max(1, len(template.gate_sequence)) * 1.5),
6,
)
per_test_agents.append(agents_per_node)
per_test_nodes.append(nodes_for_one_million)
per_test_compression.append(compression_ratio)
per_test_shots.append(int(shots_used))
domain_tokens = self._tokenize(template.domain.replace("_", " "))
if domain_tokens & prompt_tokens or domain_tokens & campaign_tokens:
domain_hits += 1
recommended_actions = list(template.playbook)
hint = self.DOMAIN_ACTION_HINTS.get(template.domain.lower())
if hint and hint not in recommended_actions:
recommended_actions.append(hint)
if template.domain.lower() == "bitcoin":
recommended_actions.append(
"Blend quantum liquidity stress with crypto telemetry to hedge exposure."
)
stress_tests.append(
QuantumStressTest(
name=template.name,
domain=template.domain,
objective=template.objective,
stressors=list(template.stressors),
metrics=list(template.metrics),
gate_sequence=list(template.gate_sequence),
measurement_probability=round(measurement, 4),
stability_index=round(stability, 4),
integration_points=list(template.integration_points),
recommended_actions=recommended_actions,
notes=template.notes,
source=capability.source,
shots_used=int(shots_used),
agents_per_node=agents_per_node,
nodes_for_one_million=nodes_for_one_million,
compression_ratio=compression_ratio,
)
)
if not stress_tests:
continue
coverage_ratio = len(coverage & self.REQUIRED_DOMAINS) / len(self.REQUIRED_DOMAINS)
avg_stability = mean(stability_scores)
score = 0.45 + coverage_ratio * 0.4 + avg_stability * 0.3 + 0.05 * domain_hits
if self.configured:
score += 0.15
if {"quantum", "simulation", "stress"} & prompt_tokens:
score += 0.08
if self.configured and coverage_ratio >= 0.8 and avg_stability >= 0.55:
maturity = "autonomous"
elif coverage_ratio >= 0.6 or avg_stability >= 0.5:
maturity = "advanced"
else:
maturity = "foundational"
domains_sorted = sorted(coverage & self.REQUIRED_DOMAINS)
reasoning_parts = []
if domains_sorted:
reasoning_parts.append(
f"Simulator covers {len(domains_sorted)} critical domains ({', '.join(domains_sorted[:5])})."
)
reasoning_parts.append(f"Average stability index {avg_stability:.2f} vs targets {mean(t.target_probability for t in capability.scenarios):.2f}.")
if self.configured:
providers = ", ".join(sorted(self._config.keys()))
reasoning_parts.append(f"Live credentials detected for {providers} environments.")
reasoning = " ".join(reasoning_parts)
orchestration_steps = list(capability.orchestration)
orchestration_steps.append(
"Distribute quantum telemetry to statistical reasoning, scenario planning, and excellence scoring."
)
orchestration_steps.append(
"Re-run stress harness after major campaign, budget, or payment changes."
)
dependencies = list(capability.dependencies)
if not self.configured:
dependencies.append(
"Configure Qiskit/AWS Braket/Google Cirq credentials to graduate to live simulations."
)
source = capability.source
if self.configured:
source = "quantum_simulation_live"
notes_parts: List[str] = []
if capability.notes:
notes_parts.append(capability.notes)
if coverage_ratio < 1.0:
missing = sorted(self.REQUIRED_DOMAINS - coverage)
if missing:
notes_parts.append(
"Additional coverage recommended for "
+ ", ".join(missing[:3])
+ ("..." if len(missing) > 3 else "")
)
if avg_stability < 0.45:
notes_parts.append("Consider increasing simulator shots or refining gate depth for stability.")
plan_agents_per_node = min(per_test_agents) if per_test_agents else 1
plan_nodes_for_million = max(per_test_nodes) if per_test_nodes else 1
plan_compression_index = (
round(mean(per_test_compression), 4) if per_test_compression else 0.0
)
plan_shots = max(per_test_shots) if per_test_shots else getattr(
self._simulator, "shots", 1024
)
continuous_operation = all(score >= 0.5 for score in stability_scores)
plan = QuantumStressTestPlan(
simulator=capability.simulator,
provider=capability.provider,
maturity=maturity,
description=capability.description,
reasoning=reasoning,
training_focus=list(capability.training_focus),
stress_tests=stress_tests,
orchestration_steps=orchestration_steps,
metrics=list(capability.evaluation_metrics or ["stability_index"]),
dependencies=dependencies,
docs=list(capability.docs),
coverage_ratio=round(coverage_ratio, 4),
domain_stability=domain_stability_map,
agents_per_node=plan_agents_per_node,
nodes_for_one_million=plan_nodes_for_million,
compression_index=plan_compression_index,
shots=int(plan_shots),
continuous_operation=continuous_operation,
source=source,
notes=" | ".join(notes_parts) if notes_parts else None,
)
plans.append((score, plan))
plans.sort(key=lambda item: item[0], reverse=True)
return [plan for _, plan in plans[: max(1, limit)]]
@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)