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.
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)]]
class LanguageIntelligenceProcess:
"""Plans how the ANI operationalizes NLP, NLU, NLG, and conversational AI."""
OPTIONAL_ENV_VARS = {
"OPENAI_API_KEY": "[credential removed]",
"ANTHROPIC_API_KEY": "[credential removed]",
"AZURE_OPENAI_ENDPOINT": "llm_providers",
"GOOGLE_API_KEY": "[credential removed]",
"GOOGLE_APPLICATION_CREDENTIALS": "language_tooling",
"AWS_REGION": "aws_region",
"AWS_ACCESS_KEY_ID": "aws_access",
"AWS_SECRET_ACCESS_KEY": "[credential removed]",
}
def __init__(
self,
capabilities: Sequence[LanguageCapability] | None = None,
dataset_path: str = LANGUAGE_CAPABILITIES_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, set()).add(env_var.lower())
return {key: sorted(values) for key, values in config.items()}
@property
def configured(self) -> bool:
return bool(self._config)
@staticmethod
def _load(path: str) -> List[LanguageCapability]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
catalog: List[LanguageCapability] = []
if isinstance(payload, list):
for entry in payload:
if isinstance(entry, dict):
try:
catalog.append(LanguageCapability.from_dict(entry))
except (KeyError, TypeError, ValueError):
continue
if not catalog:
catalog.append(
LanguageCapability(
name="Foundational Language Processing",
category="nlp",
description="Default NLP layer to tokenize and structure campaign briefs when dataset missing.",
signal_keywords=["language", "text", "prompt"],
pipeline_components=[
"Tokenization",
"Entity extraction",
"Sentiment scoring",
],
tooling=["spaCy"],
datasets=["campaign_briefs"],
evaluation_metrics=["precision", "recall"],
revenue_programs=["Audience segmentation"],
orchestration_play="Stand up baseline NLP preprocessing to unlock downstream orchestration.",
)
)
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[LanguageCapabilityPlan]:
if not campaigns:
raise ValueError("Language plans require campaign scores")
prompt_tokens = set(self._tokenize(prompt))
campaign_tokens = set(
token
for campaign in campaigns
for token in self._tokenize(
" ".join(
[
campaign.campaign.channel,
campaign.campaign.audience,
campaign.campaign.offer,
campaign.campaign.objective,
]
)
)
)
avg_confidence = mean(score.metrics.confidence for score in campaigns)
plans: List[Tuple[float, LanguageCapabilityPlan]] = []
for capability in self._capabilities:
prompt_hits = [kw for kw in capability.signal_keywords if kw in prompt_tokens]
campaign_hits = [kw for kw in capability.signal_keywords if kw in campaign_tokens]
match_score = len(prompt_hits) * 2 + len(campaign_hits)
if any(keyword in prompt.lower() for keyword in capability.signal_keywords):
match_score += 1
if capability.category in {"conversational_ai", "nlg"} and avg_confidence < 0.6:
match_score += 1
if match_score >= 6 and self.configured:
maturity = "autonomous"
elif match_score >= 3:
maturity = "advanced"
else:
maturity = "foundational"
reasoning_parts = []
if prompt_hits:
reasoning_parts.append(
f"Prompt references {', '.join(prompt_hits[:3])} signals"
)
if campaign_hits:
reasoning_parts.append(
f"Campaign corpus echoes {', '.join(campaign_hits[:3])} patterns"
)
if self.configured:
providers = ", ".join(sorted(self._config.keys()))
reasoning_parts.append(f"Environment exposes {providers} integrations")
if not reasoning_parts:
reasoning_parts.append("Default language workflow ensures briefs stay structured")
reasoning = "; ".join(reasoning_parts) + "."
orchestration_steps = [capability.orchestration_play]
orchestration_steps.extend(
[f"Operationalize {component.lower()}" for component in capability.pipeline_components[:2]]
)
if capability.tooling:
orchestration_steps.append(
f"Pair with {capability.tooling[0]} to activate production pipeline"
)
orchestration_steps.append(
"Measure "
+ ", ".join(capability.evaluation_metrics[:2] or ["precision", "recall"])
+ " across live transcripts."
)
notes: str | None = None
if self._config:
providers = sorted(
{provider for values in self._config.values() for provider in values}
)
notes = "Configured providers: " + ", ".join(providers)
plans.append(
(
match_score,
LanguageCapabilityPlan(
capability=capability.name,
category=capability.category,
maturity=maturity,
description=capability.description,
reasoning=reasoning,
pipeline_components=list(capability.pipeline_components),
tooling=list(capability.tooling),
datasets=list(capability.datasets),
evaluation_metrics=list(capability.evaluation_metrics),
revenue_programs=list(capability.revenue_programs),
orchestration_steps=orchestration_steps,
source="language_capability_catalog",
notes=notes,
),
)
)
plans.sort(key=lambda item: item[0], reverse=True)
return [plan for _, plan in plans[: max(1, limit)]]
class CodeAutomationProcess:
"""Plans how the ANI deploys AI code fixing and intelligent automation."""
OPTIONAL_ENV_VARS = {
"GITHUB_TOKEN": "[credential removed]",
"GITLAB_TOKEN": "source_control",
"BITBUCKET_USERNAME": "source_control",
"BITBUCKET_APP_PASSWORD": "[credential removed]",
"SONAR_TOKEN": "quality",
"SONAR_HOST_URL": "quality",
"SONARCLOUD_TOKEN": "quality",
"AZURE_DEVOPS_ORG_SERVICE_URL": "pipeline",
"AZURE_DEVOPS_EXT_PAT": "pipeline",
"CI": "automation",
}
def __init__(
self,
capabilities: Sequence[CodeAutomationCapability] | None = None,
dataset_path: str = CODE_AUTOMATION_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, set[str]] = {}
for env_var, key in cls.OPTIONAL_ENV_VARS.items():
value = os.getenv(env_var)
if value:
config.setdefault(key, set()).add(env_var.lower())
return {key: sorted(values) for key, values in config.items()}
@property
def configured(self) -> bool:
return bool(self._config)
@staticmethod
def _load(path: str) -> List[CodeAutomationCapability]:
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payload = []
catalog: List[CodeAutomationCapability] = []
if isinstance(payload, list):
for entry in payload:
if isinstance(entry, dict):
try:
catalog.append(CodeAutomationCapability.from_dict(entry))
except (KeyError, TypeError, ValueError):
continue
if not catalog:
catalog.append(
CodeAutomationCapability(
name="Baseline Code Automation",
category="ai_codefix",
description="Fallback automation loop that lint fixes and reruns CI when catalog missing.",
signal_keywords=["code", "fix", "automation"],
automation_stages=[
"Scan repositories",
"Generate remediation plan",
"Run unit tests",
"Deploy with guardrails",
],
tooling=["Static analyzer"],
platforms=["GitHub"],
metrics=["mean_time_to_repair"],
revenue_programs=["Engineering efficiency"],
guardrails=["Human review"],
orchestration_play="Enable baseline linting and CI automation to unblock releases.",
base_maturity="foundational",
)
)
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[CodeAutomationPlan]:
if not campaigns:
raise ValueError("Code automation plans require campaign scores")
prompt_tokens = set(self._tokenize(prompt))
campaign_tokens = set(
token
for campaign in campaigns
for token in self._tokenize(
" ".join(
[
campaign.campaign.channel,
campaign.campaign.offer,
campaign.campaign.objective,
]
)
)
)
avg_roi = mean(score.metrics.expected_roi for score in campaigns)
avg_confidence = mean(score.metrics.confidence for score in campaigns)
config_bonus = min(0.2, 0.05 * sum(len(values) for values in self._config.values()))
plans: List[Tuple[float, CodeAutomationPlan]] = []
for capability in self._capabilities:
prompt_hits = [kw for kw in capability.signal_keywords if kw in prompt_tokens]
campaign_hits = [kw for kw in capability.signal_keywords if kw in campaign_tokens]
automation_hits = [
stage
for stage in capability.automation_stages
if any(token in stage.lower() for token in prompt_tokens)
]
base = 0.45 if capability.base_maturity in {"advanced", "elite"} else 0.35
score = base + 0.09 * len(prompt_hits) + 0.07 * len(campaign_hits) + 0.05 * len(automation_hits)
if {"code", "engineering", "dev", "product"} & prompt_tokens and capability.category == "ai_codefix":
score += 0.08
if {"automation", "workflow", "process"} & prompt_tokens and capability.category == "intelligent_automation":
score += 0.08
if {"agent", "agentic", "autonomous"} & prompt_tokens and capability.category == "agentic_automation":
score += 0.08
if avg_roi >= 2.5:
score += 0.04
if avg_confidence >= 0.65:
score += 0.03
score += config_bonus
if score >= 1.05:
maturity = "autonomous"
elif score >= 0.8:
maturity = "advanced"
else:
maturity = "foundational"
base_maturity = capability.base_maturity
maturity_order = ["foundational", "advanced", "elite", "autonomous"]
if maturity_order.index(maturity) < maturity_order.index(base_maturity):
maturity = base_maturity
reasoning_parts: List[str] = []
if prompt_hits:
reasoning_parts.append(
f"Prompt references {', '.join(prompt_hits[:3])} signals tied to {capability.name}."
)
if campaign_hits:
reasoning_parts.append(
f"Campaign context mirrors {', '.join(campaign_hits[:3])} automation focus."
)
if automation_hits:
reasoning_parts.append(
f"Automation stages already align with {', '.join(automation_hits[:2])}."
)
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 = [capability.orchestration_play]
if capability.automation_stages:
orchestration_steps.append(
"Stage automation flow: " + " → ".join(capability.automation_stages[:3])
)
if capability.guardrails:
orchestration_steps.append(
"Guardrails: " + ", ".join(capability.guardrails[:2])
)
orchestration_steps.append(
"Track metrics: " + ", ".join(capability.metrics[:2] or ["mean_time_to_repair"])
)
source = "code_automation_configured" if self.configured else "code_automation_catalog"
notes = None
if not self.configured:
notes = (
"Simulated from code automation corpus; set GITHUB_TOKEN/SONAR_TOKEN/AZURE_DEVOPS_EXT_PAT "
"to enable live remediation pipelines."
)
plans.append(
(
score,
CodeAutomationPlan(
capability=capability.name,
category=capability.category,
maturity=maturity,
description=capability.description,
reasoning=reasoning,
automation_stages=list(capability.automation_stages),
tooling=list(capability.tooling),
platforms=list(capability.platforms),
metrics=list(capability.metrics),
revenue_programs=list(capability.revenue_programs),
guardrails=list(capability.guardrails),
orchestration_steps=orchestration_steps,
source=source,
notes=notes,
),
)
)
plans.sort(key=lambda item: item[0], reverse=True)
return [plan for _, plan in plans[: max(1, limit)]]
class MetaLearningProcess:
"""Scores adaptive learning capabilities that keep the ANI improving."""
OPTIONAL_ENV_VARS = {
"RAY_ADDRESS": "reinforcement",
"SAGEMAKER_ROLE": "reinforcement",
"VERTEX_AI_PROJECT": "meta",
"MLFLOW_TRACKING_URI": "distillation",
"FEATURE_STORE_URI": "self_supervised",
"GOVERNANCE_API_TOKEN": "governance",
}
def __init__(
self,
capabilities: Sequence[MetaLearningCapability] | None = None,
dataset_path: str = META_LEARNING_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, List[str]]:
config: Dict[str, set[str]] = {}
for env_var, key in cls.OPTIONAL_ENV_VARS.items():
if os.getenv(env_var):
config.setdefault(key, set()).add(env_var.lower())
return {key: sorted(values) for key, values in config.items()}
@property
def configured(self) -> bool:
return bool(self._config)
@staticmethod
def _load(path: str) -> List[MetaLearningCapability]:
capabilities: List[MetaLearningCapability] = []
try:
with open(path, "r", encoding="utf-8") as handle:
payloads = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
payloads = []
for payload in payloads:
try:
capabilities.append(MetaLearningCapability.from_dict(payload))
except (KeyError, TypeError, ValueError):
continue
if not capabilities:
capabilities.append(
MetaLearningCapability(
name="Baseline Learning Loop",
category="reinforcement_learning",
description="Fallback self-improving loop that tracks experiment results and replays the top performer.",
signal_keywords=["learning", "loop", "optimize"],
learning_modes=["manual review"],
data_sources=["campaign telemetry"],
feedback_channels=["weekly audit"],
tooling=["spreadsheet"],
metrics=["win_rate"],
deployment_modes=["manual"],
base_maturity="foundational",
)
)
return capabilities
@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[MetaLearningPlan]:
if not campaigns:
raise ValueError("Meta learning plans require campaign scores")
prompt_tokens = set(self._tokenize(prompt))
campaign_tokens = set(
token
for campaign in campaigns
for token in self._tokenize(
" ".join(
[
campaign.campaign.channel,
campaign.campaign.offer,
campaign.campaign.objective,
]
)
)
)
avg_roi = mean(score.metrics.expected_roi for score in campaigns)
avg_confidence = mean(score.metrics.confidence for score in campaigns)
config_bonus = min(0.12, 0.03 * sum(len(values) for values in self._config.values()))
plans: List[Tuple[float, MetaLearningPlan]] = []
for capability in self._capabilities:
keyword_hits = [kw for kw in capability.signal_keywords if kw in prompt_tokens]
campaign_hits = [kw for kw in capability.signal_keywords if kw in campaign_tokens]
mode_hits = [
mode
for mode in capability.learning_modes
if any(token in mode.lower() for token in prompt_tokens)
]
data_hits = [
source
for source in capability.data_sources
if any(token in source.lower() for token in prompt_tokens)
]
base = 0.45 if capability.base_maturity in {"advanced", "elite", "autonomous"} else 0.32
score = base + 0.08 * len(keyword_hits) + 0.05 * len(campaign_hits) + 0.04 * len(mode_hits)
score += 0.03 * len(data_hits)
if {"learn", "improve", "optimize"} & prompt_tokens:
score += 0.06
if avg_roi >= 2.2:
score += 0.04
if avg_confidence >= 0.65:
score += 0.03
score += config_bonus
if score >= 1.05:
maturity = "autonomous"
elif score >= 0.8:
maturity = "advanced"
else:
maturity = "foundational"
maturity_order = ["foundational", "advanced", "elite", "autonomous"]
base_index = maturity_order.index(capability.base_maturity)
maturity_index = maturity_order.index(maturity)
if maturity_index < base_index:
maturity = capability.base_maturity
reasoning_parts: List[str] = []
if keyword_hits:
reasoning_parts.append(
f"Prompt references {', '.join(keyword_hits[:3])} signals aligned with {capability.name}."
)
if campaign_hits:
reasoning_parts.append(
f"Campaign mix echoes {', '.join(campaign_hits[:3])} learning themes."
)
if mode_hits:
reasoning_parts.append(
f"Learning modes {', '.join(mode_hits[:2])} already match initiative focus."
)
if not reasoning_parts:
reasoning_parts.append(
f"Average ROI {avg_roi:.2f}x and confidence {avg_confidence:.0%} justify activating {capability.name}."
)
orchestration_steps = [
"Learning modes: " + ", ".join(capability.learning_modes[:3] or ["weekly review"]),
"Data sources: " + ", ".join(capability.data_sources[:3] or ["campaign telemetry"]),
"Feedback channels: " + ", ".join(capability.feedback_channels[:2] or ["manual audit"]),
]