core/generator.py

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",
        ]
        # Start with broad coverage across the consultant's entitled regions.
        # Adaptive seed scoring can later move capacity toward regions and
        # families that produce platform-confirmed submissions.
        self.region_weights = {
            "USA": 1,
            "CHN": 1,
            "AMR": 1,
            "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),
                    Alpha.status != "SUBMISSION_FAILED",
                )
                .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