A continuation of Janady13/AI_Brain / services/seeder_api/intelligence.py. The complete source identity, every recorded occurrence, and linked equations are on its first page.
Source transcript · continued
Download complete source text ↓10,463 lines · 396.6 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
provider_bonus += 0.18
if expert.provider == "cj" and {"affiliate", "cj", "publisher"} & prompt_tokens:
provider_bonus += 0.16
if expert.provider == "amazon" and {"amazon", "asin", "prime"} & prompt_tokens:
provider_bonus += 0.16
score = 0.42 + 0.14 * len(keyword_hits) + 0.09 * len(specialty_hits) + provider_bonus
if avg_roi >= 2.8:
score += 0.06
if avg_confidence >= 0.7:
score += 0.05
confidence = round(min(0.95, max(0.4, score / 1.2)), 2)
scored.append((score, expert, confidence))
scored.sort(key=lambda item: item[0], reverse=True)
if not scored:
return []
top_k = min(max(2, limit), len(scored))
activations: List[ExpertActivation] = []
metrics: set[str] = set()
router_notes: List[str] = []
for score, expert, confidence in scored[:top_k]:
reason_parts: List[str] = []
matched = [kw for kw in expert.trigger_keywords if kw in prompt_tokens]
if matched:
reason_parts.append(f"Router matched tokens {', '.join(matched[:3])}.")
if expert.specialties:
reason_parts.append(f"Specialty focus: {expert.specialties[0]}.")
if not matched:
reason_parts.append("Activated for coverage balance across revenue levers.")
metrics.update(expert.metrics)
router_notes.append(f"{expert.name} confidence {confidence:.0%}")
activations.append(
ExpertActivation(
name=expert.name,
provider=expert.provider,
role=expert.specialties[0] if expert.specialties else "expert",
confidence=confidence,
reasoning=" ".join(reason_parts),
outputs=list(expert.outputs),
metrics=list(expert.metrics),
fallback=expert.fallback,
source=expert.source,
)
)
avg_conf = mean(act.confidence for act in activations)
if avg_conf >= 0.8 and len(activations) >= 3:
maturity = "autonomous"
elif avg_conf >= 0.65:
maturity = "advanced"
else:
maturity = "foundational"
summary_tokens = sorted((prompt_tokens & {"google", "stripe", "cj", "amazon", "pricing", "revenue"}))
router_summary = (
"Router matched tokens: " + ", ".join(summary_tokens) if summary_tokens else "Router balanced coverage across experts"
)
notes = " | ".join(router_notes)
plan = MixtureOfExpertsPlan(
router_summary=router_summary,
maturity=maturity,
activations=activations,
evaluation_metrics=sorted(metrics) if metrics else ["roi", "ctr"],
notes=notes,
source="moe_router",
)
return [plan]
class AgenticLoopProcess:
"""Transforms prompts into continuous agentic loop blueprints."""
def __init__(
self,
loops: Sequence[Dict[str, Any]] | None = None,
dataset_path: str = AGENTIC_LOOPS_DATASET,
) -> None:
self._raw_loops = list(loops) if loops is not None else self._load(dataset_path)
@staticmethod
def _load(path: str) -> List[Dict[str, Any]]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
if isinstance(payload, list):
return [entry for entry in payload if isinstance(entry, dict)]
return []
@staticmethod
def _tokenize(text: str) -> set[str]:
return {token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2}
def plans(
self,
prompt: str,
campaigns: Sequence["CampaignScore"],
limit: int = 1,
) -> List[AgenticLoopPlan]:
if not campaigns:
raise ValueError("Agentic loop plans require campaign scores")
if not self._raw_loops:
return []
prompt_tokens = self._tokenize(prompt)
avg_confidence = mean(score.metrics.confidence for score in campaigns)
plans: List[AgenticLoopPlan] = []
for loop in self._raw_loops[: max(1, limit)]:
trigger_keywords = {kw.lower() for kw in loop.get("trigger_keywords", [])}
matched = trigger_keywords & prompt_tokens
if avg_confidence >= 0.75 and matched:
maturity = "autonomous"
elif avg_confidence >= 0.65:
maturity = "advanced"
else:
maturity = "foundational"
reasoning_parts = []
if matched:
reasoning_parts.append(
f"Prompt references {', '.join(sorted(matched))} indicating readiness for agentic loop activation."
)
reasoning_parts.append(
f"Campaign confidence {avg_confidence:.0%} supports continuous execution without drift."
)
reasoning = " ".join(reasoning_parts)
stages: List[AgenticLoopStage] = []
for stage in loop.get("stages", []):
stages.append(
AgenticLoopStage(
stage=stage.get("stage", "stage"),
objective=stage.get("objective", ""),
actions=list(stage.get("actions", [])),
tools=list(stage.get("tools", [])),
signals=list(stage.get("signals", [])),
outputs=list(stage.get("outputs", [])),
metrics=list(stage.get("metrics", [])),
)
)
plans.append(
AgenticLoopPlan(
name=loop.get("name", "Agentic Loop"),
goal_focus=loop.get("goal_focus", "marketing"),
summary=loop.get("summary", ""),
maturity=maturity,
reasoning=reasoning,
stages=stages,
guardrails=list(loop.get("guardrails", [])),
source=loop.get("source", "agentic_loop_dataset"),
)
)
return plans
class MiniQuantumSimulator:
"""Minimal single-qubit simulator used for deterministic stress scoring."""
def __init__(self) -> None:
self._sqrt_half = 1 / math.sqrt(2)
@staticmethod
def _parse_angle(text: str) -> float:
value = text.strip().lower()
if not value:
return 0.0
multiplier = -1.0 if value.startswith("-") else 1.0
if value.startswith("-"):
value = value[1:]
if value == "pi":
return multiplier * math.pi
if value.startswith("pi/"):
try:
return multiplier * math.pi / float(value[3:])
except (TypeError, ValueError):
return 0.0
if "*" in value and "pi" in value:
left, right = value.split("*", 1)
try:
factor = float(left)
except ValueError:
factor = 1.0
if right == "pi":
return multiplier * factor * math.pi
if right.startswith("pi/"):
try:
return multiplier * factor * math.pi / float(right[3:])
except (TypeError, ValueError):
return 0.0
try:
return multiplier * float(value)
except ValueError:
return 0.0
def _matrix_for_gate(self, spec: str) -> Tuple[Tuple[complex, complex], Tuple[complex, complex]]:
gate = spec.strip().upper()
identity = ((1 + 0j, 0 + 0j), (0 + 0j, 1 + 0j))
if not gate:
return identity
if gate == "H":
s = self._sqrt_half
return ((s, s), (s, -s))
if gate == "X":
return ((0 + 0j, 1 + 0j), (1 + 0j, 0 + 0j))
if gate == "Y":
return ((0 + 0j, -1j), (1j, 0 + 0j))
if gate == "Z":
return ((1 + 0j, 0 + 0j), (0 + 0j, -1 + 0j))
if gate.startswith("RX(") and gate.endswith(")"):
angle = self._parse_angle(gate[3:-1])
c = math.cos(angle / 2.0)
s = math.sin(angle / 2.0)
return ((c, -1j * s), (-1j * s, c))
if gate.startswith("RY(") and gate.endswith(")"):
angle = self._parse_angle(gate[3:-1])
c = math.cos(angle / 2.0)
s = math.sin(angle / 2.0)
return ((c, -s), (s, c))
if gate.startswith("RZ(") and gate.endswith(")"):
angle = self._parse_angle(gate[3:-1])
half = angle / 2.0
return (
(complex(math.cos(-half), math.sin(-half)), 0 + 0j),
(0 + 0j, complex(math.cos(half), math.sin(half))),
)
if gate.startswith("PHASE(") and gate.endswith(")"):
angle = self._parse_angle(gate[6:-1])
return ((1 + 0j, 0 + 0j), (0 + 0j, complex(math.cos(angle), math.sin(angle))))
return identity
def run(self, gate_sequence: Sequence[str]) -> float:
state = (1 + 0j, 0 + 0j)
for gate in gate_sequence:
matrix = self._matrix_for_gate(gate)
state = (
matrix[0][0] * state[0] + matrix[0][1] * state[1],
matrix[1][0] * state[0] + matrix[1][1] * state[1],
)
probability_one = abs(state[1]) ** 2
return float(max(0.0, min(1.0, probability_one)))
@property
def available(self) -> bool: # pragma: no cover - simple property
return True
class QuantumSimulationProcess:
"""Generates quantum stress testing plans spanning revenue systems."""
REQUIRED_DOMAINS: set[str] = {
"google_ads",
"aws",
"marketing",
"bitcoin",
"stockmarket",
"stripe",
"cashapp",
"venmo",
"paypal",
"agentic",
}
ENV_HINTS: Dict[str, List[str]] = {
"qiskit": ["QISKIT_API_TOKEN", "QISKIT_RUNTIME_API_KEY"],
"aws_braket": ["AWS_BRAKET_REGION", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"],
"google_cirq": ["GOOGLE_CIRQ_PROJECT", "GOOGLE_APPLICATION_CREDENTIALS"],
"azure_quantum": ["AZURE_QUANTUM_WORKSPACE", "AZURE_QUANTUM_RESOURCE_ID"],
}
DOMAIN_ACTION_HINTS: Dict[str, str] = {
"google_ads": "Sync quantum budget insights with Google Ads enrichment to guard ROAS.",
"aws": "Blend simulator outputs with AWS revenue playbooks for quota planning.",
"marketing": "Route funnel drift alerts into automation sprints and learning loops.",
"bitcoin": "Pair quantum volatility scores with crypto market fallbacks.",
"stockmarket": "Feed equity whipsaw metrics into scenario planning envelopes.",
"stripe": "Stage redundant checkouts when Stripe latency probability climbs.",
"cashapp": "Coordinate payout pacing with Cash App Pay readiness guides.",
"venmo": "Tune social commerce messaging using Venmo stress telemetry.",
"paypal": "Adjust PayPal reserve buffers based on quantum payout drag signals.",
"agentic": "Throttle agent workloads when loop saturation probabilities spike.",
catalog: AmazonMarketplaceCatalog | None = None,
config: Dict[str, Any] | None = None,
) -> None:
self._catalog = catalog or AmazonMarketplaceCatalog()
self._config = config or self._config_from_env()
self._session = self._build_session()
def _build_session(self) -> Any | None:
if not self.configured or boto3 is None:
return None
try: # pragma: no cover - requires boto3 dependency
return boto3.session.Session(
aws_access_key_id=self._config.get("access_key"),
aws_secret_access_key=[credential removed]"secret_key"),
region_name=self._config.get("region"),
profile_name=self._config.get("profile"),
)
except Exception:
return None
@classmethod
def _config_from_env(cls) -> Dict[str, Any]:
config: Dict[str, Any] = {}
for env_var, key in {**cls.REQUIRED_ENV_VARS, **cls.OPTIONAL_ENV_VARS}.items():
value = os.getenv(env_var)
if value:
config[key] = value
return config
@property
def configured(self) -> bool:
required = {key for key in self.REQUIRED_ENV_VARS.values()}
return required <= self._config.keys()
def listings(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int,
) -> List[AmazonProductRecommendation]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
focus = None
if campaigns:
focus = campaigns[0].campaign.channel.split()[0]
matches = self._catalog.match_listings(tokens, limit, focus)
if not matches:
return []
avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.1
avg_ctr = mean([score.metrics.expected_ctr for score in campaigns]) if campaigns else 0.025
avg_confidence = (
mean([score.metrics.confidence for score in campaigns]) if campaigns else 0.55
)
recommendations: List[AmazonProductRecommendation] = []
for match_score, listing in matches:
keyword_overlap = sorted(
{
token
for token in tokens
if token in {kw.lower() for kw in listing.keywords}
or token in listing.title.lower()
}
)
intensity = max(0.1, min(2.0, match_score / 4.0))
volume_estimate = listing.estimated_monthly_sales * max(0.4, avg_roi / 2.5)
buy_box_lift = 0.75 + listing.buy_box_share * 0.6
prime_lift = 1.18 if listing.prime_eligible else 0.9
confidence_lift = 0.85 + avg_confidence * 0.3
intent_multiplier = 0.35 + intensity * 0.25 + avg_ctr * 9.5
projected_revenue = round(
listing.price_usd
* listing.commission_rate
* volume_estimate
* buy_box_lift
* prime_lift
* confidence_lift
* intent_multiplier
/ 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,
estimated_monthly_sales=listing.estimated_monthly_sales,
buy_box_share=listing.buy_box_share,
prime_eligible=listing.prime_eligible,
fulfillment_model=listing.fulfillment_model,
projected_monthly_revenue=projected_revenue,
rationale=rationale,
url=listing.url,
source=source,
notes=notes,
)
)
return recommendations
@dataclass(slots=True)
class EbayMarketplaceListing:
"""eBay marketplace listing metadata for revenue planning."""
item_id: str
title: str
category: str
price_usd: float
affiliate_rate: float
seller_level: str
seller_rating: float
conversion_rate: float
sold_per_month: int
shipping_days: int
playbook: List[str]
confidence: float
@dataclass(slots=True)
class CJAffiliateOffer:
"""Offer metadata sourced from the Commission Junction catalog."""
program_name: str
advertiser_id: str
category: str
commission_rate: float
epc: float
seven_day_epc: float
network_rank: int
payout_terms: str
landing_url: str
description: str
keywords: List[str]
@dataclass(slots=True)
class CJAffiliateRecommendation:
"""Recommendation returned by the CJ affiliate process."""
program_name: str
advertiser_id: str
category: str
commission_rate: float
epc: float
seven_day_epc: float
projected_epc: float
projected_monthly_revenue: float
network_rank: int
payout_terms: str
landing_url: str
rationale: str
source: str
notes: str | None = None
@dataclass(slots=True)
class AmazonMarketplaceListing:
"""Amazon or AWS Marketplace listing metadata."""
asin: str
title: str
category: str
price_usd: float
commission_rate: float
rating: float
reviews: int
url: str
keywords: List[str]
@dataclass(slots=True)
class EbayRevenuePlay:
"""Recommendation payload produced by the eBay marketplace process."""
item_id: str
title: str
category: str
price_usd: float
affiliate_rate: float
seller_level: str
seller_rating: float
conversion_rate: float
sold_per_month: int
projected_monthly_revenue: float
projected_profit_margin: float
shipping_days: int
class AmazonProductRecommendation:
"""Recommendation payload produced by the Amazon marketplace process."""
asin: str
title: str
category: str
price_usd: float
commission_rate: float
rating: float
reviews: int
projected_monthly_revenue: float
rationale: str
url: str
source: str
notes: str | None = None
class EbayMarketplaceCatalog:
"""Loads curated eBay marketplace listings for offline use."""
def __init__(self, listings: Sequence[EbayMarketplaceListing] | None = None):
if listings is not None:
self._listings = list(listings)
else:
with open(EBAY_MARKETPLACE_DATASET, "r", encoding="utf-8") as handle:
data = json.load(handle)
self._listings = [EbayMarketplaceListing(**entry) for entry in data]
def match_listings(
self, tokens: Sequence[str], limit: int, focus: str | None = None
) -> List[Tuple[float, EbayMarketplaceListing]]:
ranked: List[Tuple[float, EbayMarketplaceListing]] = []
token_set = {token.lower() for token in tokens}
for listing in self._listings:
haystack = (
f"{listing.title} {listing.category} {' '.join(listing.keywords)}"
).lower()
class AmazonMarketplaceCatalog:
"""Loads curated Amazon and AWS marketplace listings for offline use."""
def __init__(self, listings: Sequence[AmazonMarketplaceListing] | None = None):
if listings is not None:
self._listings = list(listings)
else:
with open(AMAZON_MARKETPLACE_DATASET, "r", encoding="utf-8") as handle:
data = json.load(handle)
self._listings = [AmazonMarketplaceListing(**entry) for entry in data]
def match_listings(
self, tokens: Sequence[str], limit: int, focus: str | None = None
) -> List[Tuple[float, AmazonMarketplaceListing]]:
ranked: List[Tuple[float, AmazonMarketplaceListing]] = []
token_set = {token.lower() for token in tokens}
for listing in self._listings:
haystack = f"{listing.title} {listing.category} {' '.join(listing.keywords)}".lower()
score = 0.0
keyword_hits = 0
for token in token_set:
if token in haystack:
score += 1.4
keyword_hits += 1
if token in {keyword.lower() for keyword in listing.keywords}:
score += 1.6
if focus:
focus_lower = focus.lower()
if focus_lower in listing.category.lower():
score += 1.0
if focus_lower in haystack:
score += 0.6
score += min(listing.seller_rating / 5.0, 1.0)
score += min(listing.conversion_rate * 80.0, 1.2)
score += min(listing.sold_per_month / 350.0, 1.2)
if listing.seller_level.lower().startswith("top"):
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,
capabilities: Sequence[QuantumSimulatorCapability] | None = None,
dataset_path: str = QUANTUM_SIMULATIONS_DATASET,
simulator: MiniQuantumSimulator | None = None,
config: Dict[str, List[str]] | None = None,
) -> None:
self._capabilities = (
list(capabilities)
if capabilities is not None
else self._load(dataset_path)
)
self._simulator = simulator or MiniQuantumSimulator()
self._config = config or self._config_from_env()
@classmethod
def _config_from_env(cls) -> Dict[str, List[str]]:
config: Dict[str, List[str]] = {}
for name, env_vars in cls.ENV_HINTS.items():
available = [env for env in env_vars if os.getenv(env)]
if available:
config[name] = sorted({env.lower() for env in available})
return config
@property
def configured(self) -> bool:
return bool(self._config)
@staticmethod
def _tokenize(text: str) -> set[str]:
return {token for token in re.findall(r"[a-zA-Z0-9]+", text.lower()) if len(token) > 2}
@classmethod
def _load(cls, path: str) -> List[QuantumSimulatorCapability]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
capabilities: List[QuantumSimulatorCapability] = []
if isinstance(payload, list):
for entry in payload:
if not isinstance(entry, dict):
continue
scenarios: List[QuantumScenarioTemplate] = []
for raw in entry.get("scenarios", []):
if not isinstance(raw, dict):
continue
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] = []
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)
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,
)
)
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 = 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,
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)]]
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,