import json import aiohttp import random import logging import asyncio import re import time from typing import Dict, Any, List, Optional from sqlalchemy import and_, or_, select from sqlalchemy.exc import IntegrityError # Clients try: from openai import AsyncOpenAI except ImportError: AsyncOpenAI = None try: from anthropic import AsyncAnthropic except ImportError: AsyncAnthropic = None import config from database.db import AsyncSessionLocal from database.models import Alpha from utils.ingest_docs import compile_knowledge_context from core.prompts import get_system_prompt, get_user_prompt from core.expression_factory import ExpressionFactory logger = logging.getLogger("AlphaGenerator") def normalize_ast(expr: str) -> str: """ Normalizes a WorldQuant expression to its AST profile. Replaces all custom fields and constants to prevent structural duplicate submissions. """ if not expr: return "" # 1. Normalize whitespaces expr = re.sub(r'\s+', ' ', expr).strip() # 2. Match and replace numbers (scientific and regular float/int) # e.g., 10, 0.0001, 1e-5 expr = re.sub(r'\b\d+(?:\.\d+)?(?:[eE][+-]?\d+)?\b', 'N', expr) # 3. Match and replace data fields. words = re.findall(r'\b[a-zA-Z_][a-zA-Z0-9_]*\b', expr) grouping_fields = {"country", "industry", "subindustry", "currency", "market", "sector", "exchange"} reserved = {"ts_corr", "ts_covariance", "ts_mean", "ts_std_dev", "ts_decay_linear", "ts_delta", "ts_rank", "ts_zscore", "rank", "zscore", "group_rank", "group_zscore", "group_neutralize", "signed_power", "if_else", "log", "abs", "sqrt", "power", "min", "max", "sign", "exp", "ts_backfill", "group_backfill"} words_to_replace = [] for w in sorted(set(words), key=len, reverse=True): w_lower = w.lower() if w_lower in grouping_fields or w_lower in reserved: continue if w_lower in {"nan", "nil", "null", "true", "false", "n"}: continue if re.search(rf'\b{w}\b\s*\(', expr): continue words_to_replace.append(w) for w in words_to_replace: expr = re.sub(rf'\b{w}\b', 'X', expr) # Standardize spaces around operators expr = re.sub(rf'\s*([(),*+-/])\s*', r'\1', expr) return expr def research_family_key(expr: str) -> str: """Identify a formula family while retaining its data fields. ``normalize_ast`` intentionally removes field names for correlation controls. That is too broad for research selection: two formulas with different fields must not share an experimental score merely because they use the same operators. This key varies only numeric parameters. """ expr = re.sub(r"\s+", " ", expr or "").strip().lower() expr = re.sub(r"\s*([(),*+\-/])\s*", r"\1", expr) return re.sub(r"\b\d+(?:\.\d+)?(?:[eE][+-]?\d+)?\b", "N", expr) class AlphaGenerator: """ Agentic Alpha Generation Orchestrator for QuantForge v2.0. Autonomously hypothesizes, prompts, generates, validates, and queues Alpha candidates. """ def __init__(self) -> None: self.docs_context = compile_knowledge_context() self.expression_factory = ExpressionFactory() self._adaptive_seed_pool: List[Alpha] = [] self._adaptive_seed_weights: List[float] = [] self._adaptive_seed_refresh_at = 0.0 self.strategies = [ # Reversal / Mean Reversion "Short-term Price-Volume Divergence with Liquidity Shocks", "VWAP Deviation and Volume-Weighted Reversion", "Intraday Price Range Compression Mean Reversion", "Close-to-VWAP Spread Decay and Reversal", "High-Low Range Breakout Failure Reversion", # Momentum / Trend "Cross-sectional Volume Velocity and Price Momentum Decay", "Intraday Open-to-Close Spread Reversal", "High Turnover Momentum with Volatility Scaling", "Price Acceleration Signal via Second-Order Differences", "Short-Window Returns Momentum with ADV Scaling", # Liquidity / Microstructure "Bid-Ask Proxy via Close-VWAP Spread Neutralization", "Liquidity-Adjusted Returns Dispersion Signal", "Volume Shock and Price Gap Reversal", "Turnover-Adjusted Volatility Skewness Signal", "ADV20 Ratio Regime Signal with Cross-Sectional Ranking", # Statistical / Structural "Correlation Breakdown between Price and Volume Trends", "Cross-sectional Dispersion of Rolling Covariance Signals", "Rank Interaction of Price Spread and Volume Change", "Signed Power Transformation of Normalized Spread", "Decay-Weighted Composite of Open-High-Low Relationships", # Sector / Relative Value "Sector-Relative Price Strength with Volume Confirmation", "Industry-Neutral Liquidity Flow Score", "Within-Sector VWAP Deviation Ranking", "Market-Neutral ADV-Weighted Spread Signal", "Subindustry Covariance Cluster Signal", ] # Outer structural templates injected as diversity seeds self.outer_structures = [ "rank(A) - rank(B)", "rank(A) * rank(B)", "rank(A) + 0.5 * rank(B)", "rank(A) / (rank(B) + 0.001)", "signed_power(rank(A), 0.5) * rank(B)", "rank(A) - 0.5 * rank(B) * rank(C)", "rank(A * B)", "rank(A) * sign(B)", "zscore(A) * rank(B)", "rank(A) + rank(B) - rank(C)", ] self.neutralization_themes = [ "Crowding Factors", "Statistical Risk Purging", "RAM (Risk Arbitrage Model)", "Slow + Fast Factor Neutralization", "Subindustry + Industry Double Neutralization" ] self.operator_families = [ "Family A (Correlation-Reversal): use ts_corr or ts_covariance as the primary operator", "Family B (Nonlinear-Power): use signed_power with fractional exponents (e.g., signed_power(X, 0.5) or signed_power(X, 2.0))", "Family C (Momentum-Delta): use ts_delta or ts_av_diff combined with ts_zscore", "Family D (Spread-Based): use (close - open) or (close / vwap) or (high - low) or (close - low)", "Family E (Volume-Liquidity): use (volume / adv20) or ts_rank(volume, N) as primary signal", "Family F (Cross-Sectional): use group_rank or group_zscore as the outer operator", "Family G (Volatility-Deviation): use ts_std_dev or ts_zscore as the primary operator", ] # USA weight boosted: current Liquid campaign (June 29–July 12) targets USA D1 TOP1000 # CHN/AMR stay at 4x for 2.0x Dynamic Pyramid Multiplier self.region_weights = { "USA": 5, "CHN": 4, "AMR": 4, "GLB": 1, "EUR": 1, "ASI": 1, "IND": 1, "MEA": 1, } async def _refresh_adaptive_seed_pool(self) -> None: """Refresh empirically ranked seed families from persisted results.""" if time.monotonic() < self._adaptive_seed_refresh_at: return async with AsyncSessionLocal() as session: seed_result = await session.execute( select(Alpha) .where( or_( and_(Alpha.status == "SUBMITTED", Alpha.fitness >= 1.0), and_(Alpha.status == "TRASHED", Alpha.fitness >= 0.85), ), Alpha.region.in_(config.ALLOWED_REGIONS), ) .order_by(Alpha.fitness.desc(), Alpha.sharpe.desc()) .limit(160) ) seed_rows = seed_result.scalars().all() outcome_result = await session.execute( select( Alpha.expression, Alpha.region, Alpha.universe, Alpha.delay, Alpha.neutralization, Alpha.fitness, Alpha.sharpe, Alpha.turnover, ) .where( Alpha.fitness.is_not(None), Alpha.region.in_(config.ALLOWED_REGIONS), ) .order_by(Alpha.id.desc()) .limit(4000) ) outcomes = outcome_result.all() # Retain one representative per formula/configuration family. The # best near-miss can become a new local-search centre, but unrelated # data fields cannot inherit its score. def key_for(alpha: Alpha) -> tuple: return ( alpha.region, alpha.universe, alpha.delay, alpha.neutralization, research_family_key(alpha.expression), ) def threshold_for(delay: int) -> tuple[float, float]: return (2.69, 1.5) if delay == 0 else (1.58, 1.0) def quality(fitness: Any, sharpe: Any, turnover: Any, delay: int) -> float: """Score against the actual regular-alpha submission gates.""" if fitness is None or sharpe is None or turnover is None: return 0.0 target_sharpe, target_fitness = threshold_for(delay) sharpe_ratio = max(0.0, float(sharpe) / target_sharpe) fitness_ratio = max(0.0, float(fitness) / target_fitness) turnover_ok = 0.01 < float(turnover) < 0.70 return 0.55 * min(sharpe_ratio, 1.30) + 0.35 * min(fitness_ratio, 1.30) + (0.10 if turnover_ok else 0.0) best_by_family: Dict[tuple, Alpha] = {} for alpha in seed_rows: # A fitness-only near miss is exactly what dominated the prior # run. It is not a useful local-search parent without adequate # Sharpe and turnover. if alpha.status != "SUBMITTED": sharpe_target, fitness_target = threshold_for(alpha.delay) if ( alpha.sharpe is None or alpha.turnover is None or float(alpha.sharpe) < sharpe_target * 0.90 or float(alpha.fitness or 0.0) < fitness_target * 0.90 or not 0.01 < float(alpha.turnover) < 0.70 ): continue key = key_for(alpha) current = best_by_family.get(key) if current is None or quality(alpha.fitness, alpha.sharpe, alpha.turnover, alpha.delay) > quality( current.fitness, current.sharpe, current.turnover, current.delay ): best_by_family[key] = alpha observed: Dict[tuple, List[tuple[float, float, float]]] = {} for expression, region, universe, delay, neutralization, fitness, sharpe, turnover in outcomes: if fitness is None or sharpe is None or turnover is None: continue key = (region, universe, delay, neutralization, research_family_key(expression)) observed.setdefault(key, []).append((float(fitness), float(sharpe), float(turnover))) pool: List[Alpha] = [] weights: List[float] = [] retired = 0 for key, alpha in best_by_family.items(): outcomes_for_family = observed.get(key, []) scores = sorted( (quality(fitness, sharpe, turnover, alpha.delay) for fitness, sharpe, turnover in outcomes_for_family), reverse=True, )[:8] base_score = quality(alpha.fitness, alpha.sharpe, alpha.turnover, alpha.delay) if scores: best = scores[0] mean = sum(scores) / len(scores) score = 0.55 * base_score + 0.30 * best + 0.15 * mean # Retire families only after enough evidence that they miss # the real gates, not merely the old permissive local gate. if len(outcomes_for_family) >= 5 and best < 0.90: score *= 0.02 retired += 1 else: score = base_score # A retired family is not exploration; it is a known source of # wasted simulations. Remove it completely so the generator is # forced into typed or document-guided structural research. if score < 0.05: continue pool.append(alpha) weights.append(max(0.002, min(score, 1.40)) ** 6) self._adaptive_seed_pool = pool self._adaptive_seed_weights = weights self._adaptive_seed_refresh_at = time.monotonic() + 300 logger.info( "Adaptive seed pool refreshed: %s viable families from %s seed records (%s retired for weak Sharpe/fitness).", len(pool), len(seed_rows), retired, ) async def _submitted_seed_mutation(self) -> Optional[Dict[str, Any]]: """Mutate one horizon from a proven or empirically promising family. The seed's region, universe, delay, neutralization, decay and truncation remain untouched. Altering those settings turned a historical winner into a different, unvalidated experiment. """ await self._refresh_adaptive_seed_pool() seeds = self._adaptive_seed_pool if not seeds: return None seed = random.choices(seeds, weights=self._adaptive_seed_weights, k=1)[0] expression = seed.expression # The earlier three-nearest-window search was exhaustible after a # few historical runs. Use a broad, valid horizon grid and sometimes # change two independent horizons to create a genuine local search # neighbourhood rather than replaying old exact expressions. lookbacks = ( 2, 3, 4, 5, 6, 7, 8, 10, 12, 15, 18, 20, 25, 30, 35, 40, 45, 50, 60, 75, 90, 100, 120, 150, 180, 210, 252, ) horizon_pattern = "|".join(str(value) for value in sorted(lookbacks, reverse=True)) matches = list(re.finditer(rf"\b(?:{horizon_pattern})\b", expression)) if not matches: return None mutation_count = 2 if len(matches) > 1 and random.random() < 0.45 else 1 selected = random.sample(matches, k=min(mutation_count, len(matches))) for match in sorted(selected, key=lambda item: item.start(), reverse=True): current = int(match.group()) replacement = str(random.choice([value for value in lookbacks if value != current])) expression = expression[:match.start()] + replacement + expression[match.end():] return { "hypothesis": f"Adaptive one-horizon search from empirical seed Alpha {seed.id}.", "expression": expression, "region": seed.region, "universe": seed.universe, "delay": seed.delay, "decay": seed.decay, "truncation": seed.truncation, "neutralization": seed.neutralization, "language": seed.language, } async def generate_with_fallback( self, prompt: str, system_prompt: Optional[str] = None, user_prompt: Optional[str] = None ) -> str: """ Iterates over a strict priority list of 12 models from multiple providers using raw aiohttp ClientSession calls for high throughput and robustness. """ if not system_prompt: system_prompt = "You are a quantitative finance expert. Generate alpha expressions." if not user_prompt: user_prompt = prompt models_priority = [ # TIER 1: Confirmed working — Gemini lite family (separate daily quotas per model) {"model": "gemini-3.1-flash-lite", "provider": "google"}, {"model": "gemini-2.5-flash-lite", "provider": "google"}, {"model": "gemini-2.5-flash", "provider": "google"}, {"model": "gemini-3.1-pro", "provider": "google"}, # TIER 2: NIM free tier — small/fast models (<10s response) {"model": "meta/llama-3.1-8b-instruct", "provider": "nvidia"}, {"model": "deepseek-ai/deepseek-r1-distill-qwen-7b", "provider": "deepseek"}, {"model": "z-ai/glm-4-flash", "provider": "zhipu"}, # TIER 3: Last resort — slow or quota-exhausted {"model": "meta/llama-3.3-70b-instruct", "provider": "nvidia"}, {"model": "gemini-3.5-flash", "provider": "google"}, ] def get_api_key(provider: str) -> Optional[str]: if provider == "anthropic": return config.ANTHROPIC_API_KEY elif provider == "openai": return config.OPENAI_API_KEY or config.LLM_API_KEY elif provider == "google": return config.GEMINI_API_KEY or (config.GEMINI_KEYS[0] if config.GEMINI_KEYS else None) elif provider == "nvidia": return config.NVIDIA_API_KEY elif provider == "kimi": return config.MOONSHOT_API_KEY elif provider == "zhipu": return config.ZHIPU_API_KEY elif provider == "deepseek": return config.DEEPSEEK_API_KEY return None # Per-provider timeouts: NIM free tier uses short timeout to avoid 90s stalls provider_timeouts = { "nvidia": aiohttp.ClientTimeout(total=30), "deepseek": aiohttp.ClientTimeout(total=30), "zhipu": aiohttp.ClientTimeout(total=30), "google": aiohttp.ClientTimeout(total=90), "kimi": aiohttp.ClientTimeout(total=60), "anthropic":aiohttp.ClientTimeout(total=90), "openai": aiohttp.ClientTimeout(total=90), } for entry in models_priority: model = entry["model"] provider = entry["provider"] # Resolve API Keys / Key pools to try keys_to_try = [] if provider == "google": if config.GEMINI_API_KEY: keys_to_try.append(config.GEMINI_API_KEY) for k in config.GEMINI_KEYS: if k not in keys_to_try: keys_to_try.append(k) else: key = get_api_key(provider) if key: keys_to_try.append(key) # Filter out empty or whitespace keys keys_to_try = [k.strip() for k in keys_to_try if k and k.strip()] if not keys_to_try: # Key not configured, skip to next model SILENTLY continue for key in keys_to_try: try: timeout = provider_timeouts.get(provider, aiohttp.ClientTimeout(total=90)) logger.info(f"Routing request to {provider} model: {model}...") if provider == "anthropic": url = "https://api.anthropic.com/v1/messages" headers = { "x-api-key": key, "anthropic-version": "2023-06-01", "content-type": "application/json" } payload = { "model": model, "max_tokens": 1024, "messages": [ {"role": "user", "content": prompt} ] } async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post(url, json=payload, headers=headers) as response: if response.status != 200: body = await response.text() raise RuntimeError(f"HTTP Status {response.status}: {body}") res_json = await response.json() try: result_text = res_json.get("content", [{}])[0].get("text", "") except Exception as parse_err: raise ValueError(f"Failed to parse Anthropic response: {parse_err}") if not result_text: continue try: import main as _main; _main.ACTIVE_LLM_MODEL = model except Exception: pass return result_text elif provider == "google": url = f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={key}" headers = { "content-type": "application/json" } payload = { "contents": [ { "parts": [ { "text": prompt } ] } ] } async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post(url, json=payload, headers=headers) as response: if response.status != 200: body = await response.text() raise RuntimeError(f"HTTP Status {response.status}: {body}") res_json = await response.json() try: result_text = res_json.get("candidates", [{}])[0].get("content", {}).get("parts", [{}])[0].get("text", "") except Exception as parse_err: raise ValueError(f"Failed to parse Google response: {parse_err}") if not result_text: continue try: import main as _main; _main.ACTIVE_LLM_MODEL = model except Exception: pass return result_text else: # OpenAI & Compatibles if provider == "openai": url = "https://api.openai.com/v1/chat/completions" elif provider == "nvidia": url = "https://integrate.api.nvidia.com/v1/chat/completions" elif provider == "kimi": url = "https://api.moonshot.cn/v1/chat/completions" elif provider == "zhipu": if key.startswith("nvapi-"): url = "https://integrate.api.nvidia.com/v1/chat/completions" model = "z-ai/glm-5.2" else: url = "https://open.bigmodel.cn/api/paas/v4/chat/completions" elif provider == "deepseek": if key.startswith("nvapi-"): url = "https://integrate.api.nvidia.com/v1/chat/completions" model = "deepseek-ai/deepseek-v4-flash" else: url = "https://api.deepseek.com/v1/chat/completions" else: raise ValueError(f"Unknown provider: {provider}") headers = { "Authorization": f"Bearer {key}", "Content-Type": "application/json" } if provider in ["openai", "nvidia", "kimi", "zhipu", "deepseek"]: payload = { "model": model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], "temperature": 0.7, "max_tokens": 1024 } else: payload = { "model": model, "messages": [ {"role": "user", "content": prompt} ], "temperature": 0.7, "max_tokens": 1024 } async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post(url, json=payload, headers=headers) as response: if response.status != 200: body = await response.text() raise RuntimeError(f"HTTP Status {response.status}: {body}") res_json = await response.json() try: message_block = res_json.get("choices", [{}])[0].get("message", {}) result_text = message_block.get("content") or message_block.get("reasoning_content") or "" except Exception as parse_err: raise ValueError(f"Failed to parse OpenAI-compatible response: {parse_err}") if not result_text: continue try: import main as _main; _main.ACTIVE_LLM_MODEL = model except Exception: pass return result_text except Exception as e: import traceback error_msg = str(e) if not error_msg: error_msg = traceback.format_exc() logger.warning(f"Model {model} ({provider}) failed: {error_msg}. Trying next fallback...") continue # If we got here, all 12 models failed, trigger fallback logger.warning("All 12 models in the priority list failed. Triggering local deterministic fallback.") return self._generate_fallback_json() async def _call_llm(self, system_prompt: str, user_prompt: str) -> str: """ Invokes the universal multi-provider fallback router. """ combined_prompt = f"{system_prompt}\n\n{user_prompt}" return await self.generate_with_fallback( combined_prompt, system_prompt=system_prompt, user_prompt=user_prompt ) def _generate_fallback_json(self) -> str: """Generates a mock JSON string representing a valid Alpha for offline testing.""" # Randomly choose parameters to mock a correct response region = random.choice(config.ALLOWED_REGIONS) cfg = config.VALID_CONFIGS[region] universe = random.choice(cfg["universes"]) delay = random.choice(cfg["delays"]) neutralization = random.choice(cfg["neutralizations"]) grouping = neutralization.lower() if neutralization.strip().upper() != "NONE" else "subindustry" # Three exact safe templates using Price-Volume fields: raw_templates = [ f"rank(ts_decay_linear(returns, {random.randint(5, 20)})) * -1 * rank(ts_corr(close, volume, {random.randint(10, 30)}))", f"rank(ts_mean(close - vwap, {random.randint(3, 15)})) / (ts_std_dev(returns, {random.randint(20, 60)}) + 0.001)", f"rank(ts_delta(close, {random.randint(1, 5)})) * -1 * rank(ts_rank(volume, {random.randint(10, 40)}))" ] selected_raw = random.choice(raw_templates) if neutralization.strip().upper() == "NONE": expression = selected_raw else: expression = f"group_neutralize({selected_raw}, {grouping})" fallback_data = { "hypothesis": "Offline fallback testing momentum difference using PV-safe variables.", "expression": expression, "region": region, "universe": universe, "delay": delay, "decay": 5, "truncation": 0.08, "neutralization": neutralization } return json.dumps(fallback_data) async def generate_single_candidate(self) -> Optional[Dict[str, Any]]: """ Samples targets, runs the LLM call, and attempts to parse/queue the result. Uses region weighting to prioritize CHN/AMR for Dynamic Pyramid Multiplier benefit. """ # Weighted region sampling — CHN/AMR get 4x weight for 2.0x DPM multiplier. # Do not let syntactic novelty from one seed family consume the whole # research budget: every attempt chooses a lane, with independent # structural exploration receiving the remainder. seed_lane = ( config.USE_SUBMITTED_SEED_MUTATIONS and random.random() < config.SEED_MUTATION_PROBABILITY ) if seed_lane: # Exact-expression de-duplication below makes retries safe. A # broader structural block would reject every parameter mutation # because it necessarily shares its seed's operator tree. for _ in range(12): seed_candidate = await self._submitted_seed_mutation() if seed_candidate: queued = await self.parse_and_queue(json.dumps(seed_candidate)) if queued: return queued logger.info("No novel submitted-seed mutation was available in this generation attempt.") available_regions = [r for r in config.ALLOWED_REGIONS if r in self.region_weights] weights = [self.region_weights.get(r, 1) for r in available_regions] target_region = random.choices(available_regions, weights=weights, k=1)[0] if available_regions else random.choice(config.ALLOWED_REGIONS) cfg = config.VALID_CONFIGS[target_region] # Choose a target universe valid for that region target_universe = random.choice(cfg["universes"]) target_delay = random.choice(cfg["delays"]) target_neutralization = random.choice(cfg["neutralizations"]) if config.ENABLE_PURE_POWER_POOL_SUBMISSIONS and config.STRICT_POWER_POOL_THEME: theme = config.get_current_power_pool_theme() target_region = theme["region"] target_universe = theme["universe"] target_delay = theme["delay"] valid_neuts = config.VALID_CONFIGS[target_region]["neutralizations"] if target_neutralization not in valid_neuts: target_neutralization = random.choice(valid_neuts) # The typed factory is a middle lane between seed mutation and LLM # research. Keep it probabilistic so it cannot crowd out structural # exploration when it has already saturated its small template set. factory_lane = ( config.USE_TYPED_EXPRESSION_FACTORY and random.random() < 0.35 ) if factory_lane: candidate = self.expression_factory.generate( region=target_region, universe=target_universe, delay=target_delay, neutralization=target_neutralization, ) return await self.parse_and_queue(json.dumps(candidate)) if not (config.USE_LLM_FALLBACK or config.ENABLE_DOCUMENT_GUIDED_EXPLORATION): logger.info("Document-guided exploration is disabled after seed research was exhausted.") return None logger.info("Seed/template lane skipped or exhausted; starting document-guided exploration.") target_strategy = random.choice(self.strategies) neutralization_theme = random.choice(self.neutralization_themes) # Mandate a different operator family each generation to prevent correlation clustering mandated_operator_family = random.choice(self.operator_families) outer_structure_hint = random.choice(self.outer_structures) # Build numeric variation seed: random lookback windows and field pair field_pairs = [ ("close", "volume"), ("returns", "volume"), ("close", "vwap"), ("close - vwap", "volume / adv20"), ("high - low", "volume"), ("returns", "adv20"), ("close - open", "volume"), ("close", "adv20"), ] fA, fB = random.choice(field_pairs) n1 = random.randint(3, 15) n2 = random.randint(5, 25) n3 = random.randint(10, 40) variation_seed = ( f"Use lookback windows N={n1} (primary), N={n2} (secondary), N={n3} (long-term). " f"Primary field pair: ({fA}, {fB}). " f"Structural hint (adapt freely): {outer_structure_hint}." ) system_prompt = get_system_prompt(operator_family=mandated_operator_family) user_prompt = get_user_prompt( region=target_region, universe=target_universe, strategy=target_strategy, neutralization_theme=neutralization_theme, docs_context=self.docs_context, variation_seed=variation_seed ) try: raw_response = await self._call_llm(system_prompt, user_prompt) # Strip markdown block format if LLM wrapped JSON in it clean_response = raw_response.strip() if clean_response.startswith("```"): # strip code block tags lines = clean_response.splitlines() if lines[0].startswith("```"): lines = lines[1:] if lines and lines[-1].startswith("```"): lines = lines[:-1] clean_response = "\n".join(lines).strip() return await self.parse_and_queue(clean_response) except Exception as e: logger.error(f"Error in generating single candidate: {e}") return None async def generate_alpha_candidates(self, batch_size: int = 10) -> List[Dict[str, Any]]: """ Asynchronously generates multiple alpha candidates with sequential staggering. Returns a list of successfully queued alpha records. """ sem = asyncio.Semaphore(1) async def run_staggered(i): async with sem: res = await self.generate_single_candidate() # Stagger consecutive requests by sleeping if i < batch_size - 1: await asyncio.sleep(5) return res tasks = [run_staggered(i) for i in range(batch_size)] results = await asyncio.gather(*tasks, return_exceptions=True) successfully_queued = [] for res in results: if isinstance(res, dict) and res: successfully_queued.append(res) elif isinstance(res, Exception): logger.error(f"Generation task raised exception: {res}") logger.info(f"Batch generation completed. Successfully queued {len(successfully_queued)} alphas out of {batch_size}.") return successfully_queued async def parse_and_queue(self, llm_response_json: str) -> Optional[Dict[str, Any]]: """ Parses the JSON response from the LLM, validates it against region configurations and negative constraints, and queues it in the database with PENDING_SIM status. """ try: data = json.loads(llm_response_json) except json.JSONDecodeError as e: logger.error(f"Failed to parse LLM response as JSON: {e}. Raw response snippet: {llm_response_json[:100]}") return None # 1. Structural Checks required_keys = ["hypothesis", "expression", "region", "universe", "delay", "neutralization"] for key in required_keys: if key not in data: logger.warning(f"Validation failed: Missing key '{key}' in LLM response.") return None expression = str(data["expression"]).strip() region = str(data["region"]).strip().upper() universe = str(data["universe"]).strip().upper() neutralization = str(data["neutralization"]).strip().upper() try: delay = int(data["delay"]) except (ValueError, TypeError): logger.warning(f"Validation failed: Delay '{data.get('delay')}' is not a valid integer.") return None # 2. Config Validation Checks against config.py if region not in config.VALID_CONFIGS: logger.warning(f"Validation failed: Region '{region}' is not a valid search region.") return None cfg = config.VALID_CONFIGS[region] # Validate Delay support for region if delay not in cfg["delays"]: logger.warning(f"Validation failed: Delay {delay} is not supported in region '{region}' (Supported: {cfg['delays']}).") return None # Validate Universe support for region if universe not in cfg["universes"]: logger.warning(f"Validation failed: Universe '{universe}' is not supported in region '{region}' (Supported: {cfg['universes']}).") return None # Validate Neutralization support for region if neutralization not in cfg["neutralizations"]: logger.warning(f"Validation failed: Neutralization '{neutralization}' is not supported in region '{region}' (Supported: {cfg['neutralizations']}).") return None # Validate expression is not empty if not expression: logger.warning("Validation failed: Expression string is empty.") return None # 3. Add record to SQLite database async with AsyncSessionLocal() as session: # Check for duplicate expression to prevent database constraint issues dup_check = await session.execute(select(Alpha).where(Alpha.expression == expression)) if dup_check.scalars().first() is not None: logger.info(f"Duplicate Alpha expression skipped: {expression[:50]}...") return None # Parameter mutations intentionally preserve a seed's operator # tree. The old hard AST rejection made this research mode a # no-op because every mutation matched its submitted parent. # Exact expressions remain blocked above; correlation evaluation # remains the final guard for distinct variants. candidate_ast = normalize_ast(expression) active_alphas_res = await session.execute( select(Alpha.expression).where(Alpha.status != "TRASHED") ) active_exprs = active_alphas_res.scalars().all() for active_expr in active_exprs: if normalize_ast(active_expr) == candidate_ast: logger.info(f"AST profile matches an existing alpha; retaining parameter variant for evaluation: {expression[:80]}") break new_alpha = Alpha( expression=expression, language=str(data.get("language", "FASTEXPR")), region=region, universe=universe, delay=delay, decay=int(data.get("decay", 5)), truncation=float(data.get("truncation", 0.08)), neutralization=neutralization, pasteurization=str(data.get("pasteurization", "On")), nan_handling=data.get("nan_handling"), unit_handling=data.get("unit_handling"), lookback=data.get("lookback"), test_period=data.get("test_period"), max_trade=data.get("max_trade"), max_position=data.get("max_position"), status="PENDING_GEN", # Default state for simulated pipeline queue retry_count=0 ) try: session.add(new_alpha) await session.commit() except IntegrityError: await session.rollback() logger.warning(f"Duplicate expression generated by AI. Skipping... expression: {expression[:100]}") return None # Serialize for return value alpha_dict = { "id": new_alpha.id, "expression": new_alpha.expression, "region": new_alpha.region, "universe": new_alpha.universe, "delay": new_alpha.delay, "decay": new_alpha.decay, "truncation": new_alpha.truncation, "neutralization": new_alpha.neutralization, "status": new_alpha.status } logger.info(f"Alpha successfully queued in database with ID {new_alpha.id}.") return alpha_dict