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.
"""Transforms marketplace listings into revenue plans with optional SDK enrichment."""
REQUIRED_ENV_VARS = {
"AMAZON_PAAPI_ACCESS_KEY": "access_key",
"AMAZON_PAAPI_SECRET_KEY": "[credential removed]",
"AMAZON_PAAPI_PARTNER_TAG": "partner_tag",
}
OPTIONAL_ENV_VARS = {
"AWS_REGION": "region",
"AWS_PROFILE": "profile",
}
def __init__(
self,
catalog: AmazonMarketplaceCatalog | None = None,
config: Dict[str, Any] | None = None,
) -> None:
self._catalog = catalog or AmazonMarketplaceCatalog()
self._config = config or self._config_from_env()
self._session = self._build_session()
def _build_session(self) -> Any | None:
if not self.configured or boto3 is None:
return None
try: # pragma: no cover - requires boto3 dependency
return boto3.session.Session(
aws_access_key_id=self._config.get("access_key"),
aws_secret_access_key=[credential removed]"secret_key"),
region_name=self._config.get("region"),
profile_name=self._config.get("profile"),
)
except Exception:
return None
@classmethod
def _config_from_env(cls) -> Dict[str, Any]:
config: Dict[str, Any] = {}
for env_var, key in {**cls.REQUIRED_ENV_VARS, **cls.OPTIONAL_ENV_VARS}.items():
value = os.getenv(env_var)
if value:
config[key] = value
return config
@property
def configured(self) -> bool:
required = {key for key in self.REQUIRED_ENV_VARS.values()}
return required <= self._config.keys()
def listings(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int,
) -> List[AmazonProductRecommendation]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
focus = None
if campaigns:
focus = campaigns[0].campaign.channel.split()[0]
matches = self._catalog.match_listings(tokens, limit, focus)
if not matches:
return []
avg_roi = mean([score.metrics.expected_roi for score in campaigns]) if campaigns else 2.1
avg_ctr = mean([score.metrics.expected_ctr for score in campaigns]) if campaigns else 0.025
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
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
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()
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 += 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",
}
def __init__(
self,
catalog: EbayMarketplaceCatalog | None = None,
config: Dict[str, Any] | None = None,
) -> None:
self._catalog = catalog or EbayMarketplaceCatalog()
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
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]:
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] = []
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,
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
@dataclass(slots=True)
class StockMarketRecord:
"""Static fundamentals for a global stock market or macro basket."""
symbol: str
name: str
region: str
sector_focus: str
current_price: float
avg_daily_volume: float
market_cap: float
pe_ratio: float
dividend_yield: float
volatility_30d: float
momentum_score: float
macro_signal: str
marketing_correlation: float
crypto_correlation: float | None = None
@dataclass(slots=True)
class StockMarketForecast:
"""Forecast intelligence summarizing statistical outlooks for a market."""
symbol: str
name: str
region: str
focus: str
current_price: float
trend: str
forecast_return_30d: float
forecast_return_90d: float
forecast_revenue_multiplier: float
volatility: float
confidence: float
signal_summary: str
source: str
notes: str | None = None
class CryptoMarketProcess:
"""Ingests crypto market data across public APIs with graceful fallbacks."""
ASSET_ALIASES: Dict[str, str] = {
"btc": "bitcoin",
"bitcoin": "bitcoin",
"sats": "bitcoin",
"eth": "ethereum",
"ethereum": "ethereum",
"sol": "solana",
"solana": "solana",
"matic": "polygon",
"polygon": "polygon",
"doge": "dogecoin",
"dogecoin": "dogecoin",
}
DEFAULT_ASSETS: List[str] = ["bitcoin", "ethereum", "solana"]
FALLBACK_PRICES: Dict[str, Dict[str, Any]] = {
"bitcoin": {
"price_usd": 68000.0,
"change_24h": 2.5,
"market_cap": 1.3e12,
"volume_24h": 3.1e10,
"source": "simulated",
"notes": "Use CoinGecko or CoinCap credentials for live BTC spot feeds.",
},
"ethereum": {
"price_usd": 3500.0,
"change_24h": 3.1,
"market_cap": 4.2e11,
"volume_24h": 1.8e10,
"source": "simulated",
"notes": "Hook CoinGecko or CoinCap to mirror ETH market structure.",
},
"solana": {
"price_usd": 160.0,
"change_24h": 4.8,
"market_cap": 7.2e10,
"volume_24h": 4.1e9,
"source": "simulated",
"notes": "Pair CoinGecko streams for SOL liquidity snapshots.",
},
"polygon": {
"price_usd": 0.95,
"change_24h": 1.2,
"market_cap": 8.8e9,
"volume_24h": 6.7e8,
"source": "simulated",
"notes": "Integrate CoinCap assets endpoint for live Polygon metrics.",
},
"dogecoin": {
"price_usd": 0.18,
"change_24h": 5.5,
"market_cap": 2.6e10,
"volume_24h": 9.2e8,
"source": "simulated",
"notes": "CoinGecko provides DOGE spot pricing with rate-limited tier.",
},
}
def __init__(
self,
session: Any | None = None,
timeout: float = 4.0,
allow_network: bool = True,
) -> None:
self._session = None
if allow_network and requests is not None:
self._session = session or requests.Session()
self._timeout = timeout
@property
def configured(self) -> bool:
return self._session is not None
def _normalize_asset(self, token: str) -> str | None:
return self.ASSET_ALIASES.get(token.lower())
def _infer_assets(self, prompt: str, limit: int) -> List[str]:
tokens = re.findall(r"[a-zA-Z0-9]+", prompt.lower())
assets: List[str] = []
for token in tokens:
asset = self._normalize_asset(token)
if asset and asset not in assets:
assets.append(asset)
for asset in self.DEFAULT_ASSETS:
if asset not in assets:
assets.append(asset)
return assets[: max(1, limit)]
def _request_json(
self, url: str, params: Dict[str, Any] | None = None
) -> Dict[str, Any] | List[Any] | None:
if not self._session:
return None
try: # pragma: no cover - requires network
response = self._session.get(url, params=params, timeout=self._timeout)
response.raise_for_status()
return response.json()
except Exception:
return None
def _fetch_coingecko(self, asset: str) -> Dict[str, Any] | None:
data = self._request_json(
"https://api.coingecko.com/api/v3/simple/price",
params={
"ids": asset,
"vs_currencies": "usd",
"include_24hr_change": "true",
"include_market_cap": "true",
"include_24hr_vol": "true",
},
)
if isinstance(data, dict) and asset in data:
entry = data[asset]
return {
"price_usd": float(entry.get("usd")) if entry.get("usd") is not None else None,
"change_24h": float(entry.get("usd_24h_change")) if entry.get("usd_24h_change") is not None else None,
"market_cap": float(entry.get("usd_market_cap")) if entry.get("usd_market_cap") is not None else None,
"volume_24h": float(entry.get("usd_24h_vol")) if entry.get("usd_24h_vol") is not None else None,
"source": "coingecko",
}
return None
def _fetch_coincap(self, asset: str) -> Dict[str, Any] | None:
data = self._request_json(f"https://api.coincap.io/v2/assets/{asset}")
if isinstance(data, dict) and isinstance(data.get("data"), dict):
entry = data["data"]
try:
price = float(entry.get("priceUsd")) if entry.get("priceUsd") is not None else None
change = float(entry.get("changePercent24Hr")) if entry.get("changePercent24Hr") is not None else None
market_cap = float(entry.get("marketCapUsd")) if entry.get("marketCapUsd") is not None else None
volume = float(entry.get("volumeUsd24Hr")) if entry.get("volumeUsd24Hr") is not None else None
except (TypeError, ValueError):
price = change = market_cap = volume = None
return {
"price_usd": price,
"change_24h": change,
"market_cap": market_cap,
"volume_24h": volume,
"source": "coincap",
}
return None
def _fetch_blockstream(self) -> str | None:
data = self._request_json("https://blockstream.info/api/mempool")
if isinstance(data, dict) and {"count", "vsize"} <= data.keys():
return f"Blockstream mempool: {data['count']} txs, {data['vsize']} vsize"
return None
def _fetch_blockcypher(self, asset: str) -> str | None:
if asset != "bitcoin":
return None
data = self._request_json("https://api.blockcypher.com/v1/btc/main")
if isinstance(data, dict) and {"height", "unconfirmed_count"} <= data.keys():
return (
"BlockCypher height "
f"{data['height']} with {data['unconfirmed_count']} unconfirmed tx"
)
return None
def market_data(self, prompt: str, limit: int = 3) -> List[CryptoMarketSnapshot]:
assets = self._infer_assets(prompt, limit)
snapshots: List[CryptoMarketSnapshot] = []
blockstream_note = self._fetch_blockstream()
for asset in assets:
payload = self._fetch_coingecko(asset) or self._fetch_coincap(asset)
if payload is None:
fallback = self.FALLBACK_PRICES.get(asset) or self.FALLBACK_PRICES["bitcoin"]
payload = dict(fallback)
else:
payload = dict(payload)
source = payload.pop("source", "simulated")
notes = payload.pop("notes", None)
chain_note = self._fetch_blockcypher(asset)
combined_notes = " | ".join(
note for note in [notes, blockstream_note if asset == "bitcoin" else None, chain_note] if note
)
snapshots.append(
CryptoMarketSnapshot(
asset=asset,
price_usd=payload.get("price_usd"),
change_24h=payload.get("change_24h"),
market_cap=payload.get("market_cap"),
volume_24h=payload.get("volume_24h"),
source=source,
notes=combined_notes or None,
)
)
return snapshots
class StockMarketProcess:
"""Builds global stock and bitcoin macro forecasts for marketing orchestration."""
def __init__(
self,
records: Sequence[StockMarketRecord] | None = None,
dataset_path: str = STOCK_MARKETS_DATASET,
) -> None:
self._records = list(records) if records is not None else self._load_records(dataset_path)
@staticmethod
def _load_records(path: str) -> List[StockMarketRecord]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except FileNotFoundError:
payload = []
except json.JSONDecodeError:
payload = []
records: List[StockMarketRecord] = []
for entry in payload if isinstance(payload, list) else []:
try:
records.append(StockMarketRecord(**entry))
except TypeError:
continue
if not records:
records.append(
StockMarketRecord(
symbol="SPX",
name="S&P 500 Index",
region="North America",
sector_focus="Multi-sector",
current_price=5200.0,
avg_daily_volume=2.4e9,
market_cap=40.0,
pe_ratio=21.0,
dividend_yield=0.015,
volatility_30d=0.15,
momentum_score=0.35,
macro_signal="Fallback macro resilience",
marketing_correlation=0.7,
crypto_correlation=0.4,
)
)
return records
@staticmethod
def _tokenize(text: str) -> List[str]:
return re.findall(r"[a-zA-Z0-9]+", text.lower())
def _score_record(
self,
record: StockMarketRecord,
tokens: Sequence[str],
campaigns: Sequence[CampaignScore],
) -> float:
token_score = 0.0
haystack = f"{record.name} {record.region} {record.sector_focus} {record.symbol}".lower()
for token in tokens:
if token in haystack:
token_score += 0.6
if token in record.sector_focus.lower():
token_score += 0.4
if token in record.region.lower():
token_score += 0.4
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
)
momentum_component = record.momentum_score * 2.0
correlation_component = record.marketing_correlation * 1.5
volatility_penalty = (1.0 - record.volatility_30d) * 1.2
roi_alignment = avg_roi * 0.3
confidence_alignment = avg_confidence * 0.5
return (
token_score
+ momentum_component
+ correlation_component
+ volatility_penalty
+ roi_alignment
+ confidence_alignment
)
@staticmethod
def _trend_label(record: StockMarketRecord) -> str:
if record.momentum_score >= 0.45:
return "surging"
if record.momentum_score >= 0.35:
return "climbing"
if record.momentum_score >= 0.25:
return "stabilizing"
return "range bound"
def forecasts(
self,
prompt: str,
campaigns: Sequence[CampaignScore],
limit: int = 3,
) -> List[StockMarketForecast]:
limit = max(1, limit)
tokens = self._tokenize(prompt)
ranked = sorted(
self._records,
key=lambda record: self._score_record(record, tokens, campaigns),
reverse=True,
)[:limit]
forecasts: List[StockMarketForecast] = []
avg_roi = mean(score.metrics.expected_roi for score in campaigns) if campaigns else 2.0
avg_ctr = mean(score.metrics.expected_ctr for score in campaigns) if campaigns else 0.03
for record in ranked:
volatility = max(0.05, min(0.6, record.volatility_30d))
base_return = record.momentum_score * 0.08 + record.marketing_correlation * 0.05
volatility_adjustment = (0.25 - volatility) * 0.12
thirty_day = max(-0.12, min(0.22, base_return + volatility_adjustment))
ninety_day = max(
-0.25,
min(0.45, thirty_day * 2.1 + record.momentum_score * 0.05 - volatility * 0.1),
)
revenue_multiplier = max(
0.6,
min(2.8, (avg_roi / 2.0) * (1 + thirty_day) + avg_ctr * 20),
)
confidence = max(
0.25,
min(
0.95,
0.45
+ record.marketing_correlation * 0.25
+ (1 - volatility) * 0.15
+ record.momentum_score * 0.15
+ (record.crypto_correlation or 0.3) * 0.05,
),
)
trend = self._trend_label(record)
signal_summary = (
f"{record.name} momentum {trend} with {thirty_day*100:.1f}% 30d outlook "
f"and revenue multiplier {revenue_multiplier:.2f}x"
)
notes_parts = [record.macro_signal]
if record.crypto_correlation is not None and record.crypto_correlation > 0.5:
notes_parts.append("High crypto correlation supports synchronized BTC hedges")
forecasts.append(
StockMarketForecast(
symbol=record.symbol,
name=record.name,
region=record.region,
focus=record.sector_focus,
current_price=round(record.current_price, 2),
trend=trend,
forecast_return_30d=round(thirty_day, 4),
forecast_return_90d=round(ninety_day, 4),
forecast_revenue_multiplier=round(revenue_multiplier, 2),
volatility=round(volatility, 3),
confidence=round(confidence, 2),
signal_summary=signal_summary,
source="stock_catalog",
notes=" | ".join(notes_parts),
)
)
return forecasts
class StatisticalReasoningStore:
"""Optional Postgres sink for statistical reasoning summaries."""
def __init__(self, enabled: bool | None = None) -> None:
env_flag = os.getenv("STATISTICAL_REASONING_PERSIST", "0").lower()
self._enabled = (
enabled
if enabled is not None
else env_flag in {"1", "true", "yes", "on"}
)
self._config = {
"dbname": os.getenv("PGDATABASE", os.getenv("DB_NAME", "brain_atomic")),
"user": os.getenv("PGUSER", os.getenv("DB_USER", "postgres")),
"host": os.getenv("PGHOST", os.getenv("DB_HOST", "localhost")),
"port": int(os.getenv("PGPORT", os.getenv("DB_PORT", "5432"))),
}
password = os.getenv("PGPASSWORD", os.getenv("DB_PASSWORD"))
if password:
self._config["password"] = password
@property
def enabled(self) -> bool:
return self._enabled and psycopg2 is not None
def record(self, prompt: str, summary: StatisticalReasoningSummary) -> bool:
if not self.enabled:
return False
conn = None
cur = None
try: # pragma: no cover - requires live Postgres
conn = psycopg2.connect(**self._config) # type: ignore[arg-type]
cur = conn.cursor()
if Json is not None:
spread_payload: Any = Json(summary.scenario_spread) # type: ignore[call-arg]
drivers_payload: Any = Json(summary.driver_confidence) # type: ignore[call-arg]
query = (
"INSERT INTO statistical_reasoning_insights (prompt, expected_roi, expected_revenue, "
"confidence, success_probability, volatility_adjustment, scenario_spread, drivers, narrative, notes) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
)
params = (
prompt,
summary.expected_roi,
summary.expected_revenue,
summary.confidence,
summary.success_probability,
summary.volatility_adjustment,
spread_payload,
drivers_payload,
summary.narrative,
summary.notes,
)
else:
spread_payload = json.dumps(summary.scenario_spread)
drivers_payload = json.dumps(summary.driver_confidence)
query = (
"INSERT INTO statistical_reasoning_insights (prompt, expected_roi, expected_revenue, "
"confidence, success_probability, volatility_adjustment, scenario_spread, drivers, narrative, notes) "
"VALUES (%s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s, %s)"
)
params = (
prompt,
summary.expected_roi,
summary.expected_revenue,
summary.confidence,
summary.success_probability,
summary.volatility_adjustment,
spread_payload,
drivers_payload,
summary.narrative,
summary.notes,
)
cur.execute(query, params)
conn.commit()
return True
except Exception: # pragma: no cover - persistence best-effort
return False
finally:
if cur is not None:
cur.close()
if conn is not None:
conn.close()
class StatisticalReasoningProcess:
"""Blends statistical priors with live campaign metrics to guide revenue."""
def __init__(
self,
priors: Sequence[StatisticalPrior] | None = None,
dataset_path: str = STATISTICAL_PRIORS_DATASET,
store: StatisticalReasoningStore | None = None,
) -> None:
self._priors = list(priors) if priors is not None else self._load_priors(dataset_path)
self._store = store or StatisticalReasoningStore()
@staticmethod
def _load_priors(path: str) -> List[StatisticalPrior]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
priors: List[StatisticalPrior] = []
if isinstance(payload, list):
for entry in payload:
if not isinstance(entry, dict):
continue
try:
priors.append(
StatisticalPrior(
feature=str(entry.get("feature", "unspecified")),
feature_type=str(entry.get("feature_type", "feature")),
mean_roi=float(entry.get("mean_roi", 2.0)),
std_roi=max(0.01, float(entry.get("std_roi", 0.5))),
mean_ctr=float(entry.get("mean_ctr", 0.03)),
std_ctr=max(0.0001, float(entry.get("std_ctr", 0.01))),
sample_size=int(entry.get("sample_size", 100)),
weight=float(entry.get("weight", 0.75)),
baseline_budget=float(entry.get("baseline_budget", 20000.0)),
notes=str(entry.get("notes", "Statistical prior")),
)
)
except (TypeError, ValueError):
continue
if not priors:
priors.append(
StatisticalPrior(
feature="omni-channel",
feature_type="channel",
mean_roi=2.2,
std_roi=0.5,
mean_ctr=0.032,
std_ctr=0.01,
sample_size=120,
weight=0.75,
baseline_budget=20000.0,
notes="Fallback statistical prior",
)
)
return priors
@staticmethod
def _normalize(text: str) -> str:
return re.sub(r"\s+", " ", text.lower().strip())
def _match_priors(self, value: str, feature_type: str) -> List[StatisticalPrior]:
target = self._normalize(value)
matches: List[StatisticalPrior] = []
for prior in self._priors:
if prior.feature_type != feature_type:
continue
feature = self._normalize(prior.feature)
if target == feature or target in feature or feature in target:
matches.append(prior)
return matches
def analyze(
self,
prompt: str,
campaigns: Sequence["CampaignScore"],
*,
budget: BudgetPlan | None,
stock_markets: Sequence[StockMarketForecast] | None,
crypto_markets: Sequence[CryptoMarketSnapshot] | None,
limit: int = 3,
persist: bool = True,
) -> StatisticalReasoningSummary:
if not campaigns:
raise ValueError("Statistical reasoning requires campaign scores")
top = list(campaigns[: max(1, limit)])
evidence: List[StatisticalEvidence] = []
driver_confidence: Dict[str, float] = {}
baseline_budgets: List[float] = []
for score in top:
feature_map = {
"channel": score.campaign.channel,
"audience": score.campaign.audience,
"offer": score.campaign.offer,
"objective": score.campaign.objective,
}
for feature_type, raw_value in feature_map.items():
priors = self._match_priors(raw_value, feature_type)
if not priors:
continue
prior = priors[0]
baseline_budgets.append(prior.baseline_budget)
prior_weight = max(1.0, prior.sample_size * prior.weight)
observation_weight = max(1.0, score.metrics.confidence * 120.0)
posterior_roi = (
prior.mean_roi * prior_weight + score.metrics.expected_roi * observation_weight
) / (prior_weight + observation_weight)
posterior_ctr = (
prior.mean_ctr * prior_weight + score.metrics.expected_ctr * observation_weight
) / (prior_weight + observation_weight)
lift = (posterior_roi - prior.mean_roi) / max(0.1, prior.mean_roi)
confidence = max(
0.35,
min(
0.98,
score.metrics.confidence * 0.6 + prior.weight * 0.4,
),
)
rationale = (
f"Posterior {posterior_roi:.2f}x ROI blends prior ({prior.mean_roi:.2f}x) "
f"with observed {score.metrics.expected_roi:.2f}x and {prior.sample_size} runs."
)
evidence.append(
StatisticalEvidence(
feature=prior.feature,
feature_type=prior.feature_type,
posterior_roi=round(posterior_roi, 3),
posterior_ctr=round(posterior_ctr, 4),
confidence=round(confidence, 3),
lift_vs_benchmark=round(lift, 3),
sample_size=prior.sample_size,
weight=round(prior.weight, 3),
rationale=rationale,
)
)
driver_key = f"{prior.feature_type}:{prior.feature}"
driver_confidence[driver_key] = max(
driver_confidence.get(driver_key, 0.0), confidence
)
if not evidence:
fallback_prior = self._priors[0]
evidence.append(
StatisticalEvidence(
feature=fallback_prior.feature,
feature_type=fallback_prior.feature_type,
posterior_roi=round(fallback_prior.mean_roi, 3),
posterior_ctr=round(fallback_prior.mean_ctr, 4),
confidence=0.55,
lift_vs_benchmark=0.0,
sample_size=fallback_prior.sample_size,
weight=round(fallback_prior.weight, 3),
rationale="Fallback statistical prior applied due to limited matches.",
)
)
driver_confidence[f"{fallback_prior.feature_type}:{fallback_prior.feature}"] = 0.55
baseline_budgets.append(fallback_prior.baseline_budget)
total_weight = sum(ev.weight for ev in evidence) or 1.0
expected_roi = sum(ev.posterior_roi * ev.weight for ev in evidence) / total_weight
expected_ctr = sum(ev.posterior_ctr * ev.weight for ev in evidence) / total_weight
evidence_confidence = sum(ev.confidence for ev in evidence) / len(evidence)
avg_confidence = mean(score.metrics.confidence for score in top)
roi_signal = mean(score.metrics.expected_roi for score in top)
budget_baseline = (
budget.total_budget
if budget and budget.total_budget > 0
else (sum(baseline_budgets) / len(baseline_budgets) if baseline_budgets else 22000.0)
)
expected_revenue = round(budget_baseline * max(expected_roi, roi_signal, 1.0), 2)
stock_vol = (
mean(forecast.volatility for forecast in stock_markets) if stock_markets else 0.22
)
crypto_vol = (
mean(abs(snapshot.change_24h or 0) / 100 for snapshot in crypto_markets)
if crypto_markets
else 0.12
)
volatility_adjustment = round(min(0.4, stock_vol * 0.6 + crypto_vol * 0.4), 3)
confidence = max(
0.35,
min(
0.98,
evidence_confidence * 0.5
+ avg_confidence * 0.3
+ min(0.25, expected_ctr * 5)
- volatility_adjustment * 0.35,
),
)
success_probability = max(
0.15,
min(
0.99,
0.5
+ (expected_roi - 1.0) * 0.18
+ (confidence - 0.5) * 0.5
- volatility_adjustment * 0.25,
),
)
scenario_spread = {
"optimistic": round(expected_roi * (1 + volatility_adjustment * 0.4 + 0.2), 2),
"base": round(expected_roi, 2),
"conservative": round(max(0.2, expected_roi * (1 - (volatility_adjustment * 0.5 + 0.15))), 2),
}
sorted_drivers = sorted(
driver_confidence.items(), key=lambda item: item[1], reverse=True
)
top_driver_text = (
sorted_drivers[0][0].split(":", 1)[1] if sorted_drivers else "statistical priors"
)
narrative = (
f"Statistical reasoning projects {expected_roi:.2f}x ROI with {success_probability:.0%} "
f"success probability anchored by {top_driver_text}."
)
if stock_markets:
narrative += (
f" Market volatility adjustment applied at {volatility_adjustment:.2f} to reflect equity signals."
)
driver_confidence = {
key: round(min(1.0, max(0.0, value)), 3) for key, value in driver_confidence.items()
}
notes = (
f"Blended {len(evidence)} statistical priors across {len(top)} campaigns; "
f"baseline budget {budget_baseline:,.0f}."
)
summary = StatisticalReasoningSummary(
expected_roi=round(expected_roi, 3),
expected_revenue=expected_revenue,
confidence=round(confidence, 3),
success_probability=round(success_probability, 3),
volatility_adjustment=volatility_adjustment,
scenario_spread=scenario_spread,
driver_confidence=driver_confidence,
evidence=evidence,
narrative=narrative,
notes=notes,
)
if persist:
persisted = self._store.record(prompt, summary)
if persisted:
existing = summary.notes or ""
note_parts = [part for part in [existing, "Persisted to Postgres"] if part]
summary.notes = " | ".join(note_parts)
return summary
class CognitiveArchitectureProcess:
"""Infers cognitive capabilities that power the marketing ANI."""
def __init__(
self,
capabilities: Sequence[CognitiveCapability] | None = None,
dataset_path: str = COGNITIVE_CAPABILITIES_DATASET,
) -> None:
self._capabilities = list(capabilities) if capabilities is not None else self._load(dataset_path)
@staticmethod
def _load(path: str) -> List[CognitiveCapability]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
catalog: List[CognitiveCapability] = []
if isinstance(payload, list):
for entry in payload:
if isinstance(entry, dict):
try:
catalog.append(CognitiveCapability.from_dict(entry))
except (KeyError, TypeError, ValueError):
continue
if not catalog:
catalog.append(
CognitiveCapability(
name="Foundational Cognitive Loop",
pillar="foundational",
description="Default cognitive layer ensuring reasoning and learning remain available.",
signal_keywords=["cognitive", "learning", "reasoning"],
prompt_strategies=[
"Request blended learning and reasoning prompts to keep orchestration robust."
],
frameworks=["Knowledge-Augmented Generation"],
tooling=["Vector memory"],
metrics=["excellence_index"],
knowledge_assets=["campaign_briefs"],
play="Maintain base cognitive readiness for marketing ANI.",
)
)
return catalog
@staticmethod
def _tokenize(text: str) -> List[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 = 3,
) -> List[CognitiveCapabilityPlan]:
if not campaigns:
raise ValueError("Cognitive architecture plans require campaign scores")
prompt_lower = prompt.lower()
prompt_tokens = set(self._tokenize(prompt))
campaign_corpus = " ".join(score.campaign.raw for score in campaigns).lower()
campaign_tokens = set(self._tokenize(campaign_corpus))
avg_roi = mean(score.metrics.expected_roi for score in campaigns)
avg_confidence = mean(score.metrics.confidence for score in campaigns)
plans: List[Tuple[float, CognitiveCapabilityPlan]] = []
for capability in self._capabilities:
keyword_hits = [kw for kw in capability.signal_keywords if kw in prompt_lower]
campaign_hits = [kw for kw in capability.signal_keywords if kw in campaign_corpus]
framework_hits = [
framework
for framework in capability.frameworks
if framework.lower() in prompt_lower
]
score = 0.35 + 0.12 * len(keyword_hits) + 0.08 * len(campaign_hits) + 0.07 * len(framework_hits)
if capability.pillar == "machine_learning" and avg_roi >= 2.5:
score += 0.12
if capability.pillar == "reasoning" and avg_confidence >= 0.65:
score += 0.1
if capability.pillar in {"perception", "vector_databases"} and campaign_tokens & {"amazon", "google", "stock", "crypto"}:
score += 0.09
if capability.pillar in {"prompt_engineering", "prompt_management", "language"} and "prompt" in prompt_tokens:
score += 0.08
maturity: str
if score >= 0.95:
maturity = "autonomous"
elif score >= 0.7:
maturity = "advanced"
else:
maturity = "foundational"
reasoning_parts: List[str] = []
if keyword_hits:
reasoning_parts.append(
f"Prompt references {', '.join(keyword_hits)} which align to {capability.name}."
)
if framework_hits:
reasoning_parts.append(
f"Frameworks {', '.join(framework_hits)} reinforce the capability stack."
)
if not reasoning_parts:
reasoning_parts.append(
f"Campaign metrics ({avg_roi:.2f}x ROI, {avg_confidence:.0%} confidence) warrant activating {capability.name}."
)
reasoning = " ".join(reasoning_parts)
prompt_strategy = capability.prompt_strategies[0]
if campaigns and any(
placeholder in prompt_strategy
for placeholder in ["{audience}", "{channel}", "{offer}"]
):
prompt_strategy = prompt_strategy.format(
audience=campaigns[0].campaign.audience,
channel=campaigns[0].campaign.channel,
offer=campaigns[0].campaign.offer,
)
orchestration_steps = [capability.play]
if framework_hits:
orchestration_steps.append(
f"Accelerate using {', '.join(framework_hits)} for immediate deployment."
)
else:
orchestration_steps.append(
f"Pair with {capability.frameworks[0]} to operationalize within the ANI."
)
orchestration_steps.append(
"Measure impact via " + ", ".join(capability.metrics[:2]) + "."
)
plans.append(
(
score,
CognitiveCapabilityPlan(
capability=capability.name,
pillar=capability.pillar,
maturity=maturity,
description=capability.description,
reasoning=reasoning,
prompt_strategy=prompt_strategy,
frameworks=capability.frameworks,
tooling=capability.tooling,
metrics=capability.metrics,
knowledge_assets=capability.knowledge_assets,
orchestration_steps=orchestration_steps,
source="cognitive_capability_catalog",
),
)
)
plans.sort(key=lambda item: item[0], reverse=True)
top_plans = [plan for _, plan in plans[: max(1, limit)]]
return top_plans
class AgenticSystemsProcess:
"""Plans agentic system activation across marketing, data, and trading workflows."""
OPTIONAL_ENV_VARS = {
"DATABASE_URL": "database_url",
"STATISTICS_DATABASE_URL": "statistics_db",
"FEATURE_STORE_URL": "feature_store",
"VECTOR_DB_URL": "vector_db",
"PINECONE_API_KEY": "[credential removed]",
"BIGQUERY_PROJECT": "bigquery_project",
"AWS_REGION": "aws_region",
"BEDROCK_ENDPOINT": "prompt_orchestrator",
}
def __init__(
self,
capabilities: Sequence[AgenticCapability] | None = None,
dataset_path: str = AGENTIC_SYSTEMS_DATASET,
config: Dict[str, Any] | None = None,
) -> None:
self._capabilities = list(capabilities) if capabilities is not None else self._load(dataset_path)
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.setdefault(key, value)
return config
@property
def configured(self) -> bool:
return bool(self._config)
@staticmethod
def _load(path: str) -> List[AgenticCapability]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
catalog: List[AgenticCapability] = []
if isinstance(payload, list):
for entry in payload:
if isinstance(entry, dict):
try:
catalog.append(AgenticCapability.from_dict(entry))
except (KeyError, TypeError, ValueError):
continue
if not catalog:
catalog.append(
AgenticCapability(
name="Agentic Growth Skeleton",
domain="agentic_systems",
description="Baseline agent workflow ensuring orchestration across marketing processes.",
controller="LLM executive controller",
trigger_keywords=["agent", "automation", "workflow"],
frameworks=["Prompt management"],
toolchain=["Process orchestrator"],
data_interfaces=["Marketing warehouse"],
knowledge_sources=["Campaign briefs"],
revenue_programs=["Acquisition"],
risk_controls=["Budget checkpoints"],
maturity_signals=["Automation coverage"],
base_maturity="foundational",
metrics=["agent_cycle_time"],
)
)
return catalog
@staticmethod
def _tokenize(text: str) -> List[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 = 3,
) -> List[AgenticSystemPlan]:
if not campaigns:
raise ValueError("Agentic system plans require campaign scores")
prompt_tokens = set(self._tokenize(prompt))
campaign_corpus = " ".join(score.campaign.raw for score in campaigns)
campaign_tokens = set(self._tokenize(campaign_corpus))
avg_roi = mean(score.metrics.expected_roi for score in campaigns)
avg_confidence = mean(score.metrics.confidence for score in campaigns)
config_score = min(0.2, 0.05 * len(self._config))
psycopg2_available = psycopg2 is not None
scored_plans: List[Tuple[float, AgenticSystemPlan]] = []
for capability in self._capabilities:
keyword_hits = [kw for kw in capability.trigger_keywords if kw in prompt_tokens]
campaign_hits = [kw for kw in capability.trigger_keywords if kw in campaign_tokens]
framework_hits = [
framework
for framework in capability.frameworks
if framework.lower() in prompt_tokens
]
data_hits = [
interface
for interface in capability.data_interfaces
if any(keyword in interface.lower() for keyword in prompt_tokens)
]
base = 0.45 if capability.base_maturity in {"advanced", "elite"} else 0.35
score = base + 0.09 * len(keyword_hits) + 0.07 * len(campaign_hits) + 0.06 * len(framework_hits)
score += 0.05 * len(data_hits)
score += config_score
if capability.domain in {"machine_learning", "data_connectivity"} and psycopg2_available:
score += 0.05
if capability.domain == "market_operations" and any(
token in {"trading", "market", "portfolio"} for token in prompt_tokens
):
score += 0.06
if capability.domain == "customer_experience" and any(
token in {"support", "retention", "customer"} for token in prompt_tokens
):
score += 0.05
if capability.domain == "prompt_operations" and {"prompt", "cache", "knowledge"} & prompt_tokens:
score += 0.05
if avg_roi >= 2.8:
score += 0.05
if avg_confidence >= 0.7:
score += 0.04
if score >= 1.05:
maturity = "autonomous"
elif score >= 0.85:
maturity = "elite"
elif score >= 0.65:
maturity = "advanced"
else:
maturity = "foundational"
if maturity == "foundational" and capability.base_maturity in {"advanced", "elite"}:
maturity = capability.base_maturity
reasoning_parts = []
if keyword_hits:
reasoning_parts.append(
f"Prompt signals {', '.join(keyword_hits[:3])} intent aligning to {capability.name}."
)
if framework_hits:
reasoning_parts.append(
f"Framework cues {', '.join(framework_hits[:2])} strengthen orchestration quality."
)
if data_hits:
reasoning_parts.append(
f"Data surfaces ({', '.join(data_hits[:2])}) ensure live telemetry feeds."
)
if not reasoning_parts:
reasoning_parts.append(
f"Campaign ROI {avg_roi:.2f}x and confidence {avg_confidence:.0%} justify activating {capability.name}."
)
reasoning = " ".join(reasoning_parts)
orchestration_steps = list(capability.toolchain[:3])
orchestration_steps.append(
f"Wire {capability.controller} to {capability.frameworks[0] if capability.frameworks else 'agent runner'}"
)
orchestration_steps.append("Instrument metrics: " + ", ".join(capability.metrics[:3]))
source = "agentic_systems_live" if self.configured else "agentic_systems_catalog"
notes = None
if not self.configured:
notes = (
"Simulated from agentic systems corpus; set DATABASE_URL / FEATURE_STORE_URL / VECTOR_DB_URL "
"to enable live tool calling."
)
scored_plans.append(
(
score,
AgenticSystemPlan(
system_name=capability.name,
domain=capability.domain,
controller=capability.controller,
maturity=maturity,
reasoning=reasoning,
orchestration_steps=orchestration_steps,
toolchain=list(capability.toolchain),
data_interfaces=list(capability.data_interfaces),
knowledge_sources=list(capability.knowledge_sources),
revenue_programs=list(capability.revenue_programs),
risk_controls=list(capability.risk_controls),
metrics=list(capability.metrics),
source=source,
notes=notes,
),
)
)
scored_plans.sort(key=lambda item: item[0], reverse=True)
return [plan for _, plan in scored_plans[: max(1, limit)]]