import json import logging from typing import Dict, Any, Tuple, Optional import config from database.db import AsyncSessionLocal from database.models import Alpha logger = logging.getLogger("AlphaEvaluator") def serialize_raw_payload(raw_data: Any) -> str: """Preserve completed API payloads for parser and research forensics.""" payload = json.dumps(raw_data or {}, default=str, separators=(",", ":")) return payload[:250_000] # Trash reason categories for analytics TRASH_REASON_NO_TRADES = "No Trades / Null Metrics" TRASH_REASON_LOW_PERFORMANCE = "Low Sharpe / Fitness" TRASH_REASON_HIGH_CORRELATION = "High Correlation" TRASH_REASON_SYNTAX_ERROR = "Syntax / Schema Error" def classify_trash_reason(reason: str) -> str: """ Classifies a failure reason string into one of 4 canonical trash categories for structured analytics on the dashboard. """ r = reason.lower() if "no trades" in r or "null" in r or "invalid stats" in r or "missing critical" in r: return TRASH_REASON_NO_TRADES if "sharpe" in r or "fitness" in r or "turnover" in r or "performance" in r: return TRASH_REASON_LOW_PERFORMANCE if "correlation" in r or "corr" in r: return TRASH_REASON_HIGH_CORRELATION # Simulation failures, schema errors, platform rejects, syntax errors return TRASH_REASON_SYNTAX_ERROR class AlphaEvaluator: """ Evaluation Engine for QuantForge v2.0. Applies strict Consultant-tier acceptance thresholds on simulated Alphas. """ def __init__(self) -> None: # Ordinary consultant submission thresholds. Pure Power Pool uses a # distinct campaign and is disabled until that engine exists. self.min_fitness_d1 = 1.0 self.min_fitness_d0 = 1.5 self.min_sharpe_d1 = 1.58 self.min_sharpe_d0 = 2.69 self.min_turnover = 0.01 self.max_turnover = 0.70 self.min_htvr_ratio = 0.75 # High Turnover Returns Ratio for x2 multiplier eligibility self.max_self_correlation = 0.70 self.max_prod_correlation = 0.70 def evaluate_metrics(self, sim_results: Dict[str, Any], delay: int, expression: Optional[str] = None) -> Tuple[bool, str]: """ Evaluates parsed simulation results against thresholds. Returns: (is_passing, failure_reason) """ # 1. Check for system or platform errors reported during simulation/polling if sim_results.get("status") == "FAILED": err_msg = sim_results.get("error_message") or "Simulation status failed." return False, f"Simulation failure: {err_msg}" # 2. Check for missing essential metrics sharpe = sim_results.get("sharpe") fitness = sim_results.get("fitness") turnover = sim_results.get("turnover") if sharpe is None or fitness is None or turnover is None: msg = "Simulation produced no trades / Invalid Stats" logger.warning(msg) return False, msg # 3. Sharpe cutoff — match the ordinary consultant IS thresholds. # D0: 2.69 | D1: 1.58. Do not spend submit quota on predictable rejects. min_sharpe = self.min_sharpe_d0 if delay == 0 else self.min_sharpe_d1 # Pure Power Pool is opt-in only; it has its own themed campaign rules. if expression: try: import re # Calculate complexity operators = re.findall(r'\b[a-zA-Z_][a-zA-Z0-9_]*\b(?=\s*\()', expression) filtered_ops = [op for op in operators if op.lower() not in {"ts_backfill", "group_backfill"}] num_ops = len(filtered_ops) words = re.findall(r'\b[a-zA-Z_][a-zA-Z0-9_]*\b', expression) 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"} fields = set() for w in words: 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"}: continue fields.add(w_lower) num_fields = len(fields) if config.ENABLE_PURE_POWER_POOL_SUBMISSIONS and num_fields <= 3 and num_ops <= 8: logger.info(f"Alpha complexity qualifies for Power Pool (fields={num_fields}, ops={num_ops}). Setting min_sharpe = 1.0.") min_sharpe = 1.0 except Exception as e: logger.error(f"Error checking Power Pool complexity: {e}") if sharpe <= min_sharpe: return False, f"Sharpe {sharpe:.4f} below pre-filter threshold of {min_sharpe} (Delay {delay})" # 4. Fitness Cutoff min_fitness = self.min_fitness_d0 if delay == 0 else self.min_fitness_d1 if fitness <= min_fitness: return False, f"Fitness {fitness:.4f} below threshold of {min_fitness}" # 5. Turnover band for regular consultant alphas. if turnover <= self.min_turnover: return False, f"Turnover {turnover:.4f} below floor of {self.min_turnover}" if turnover >= self.max_turnover: return False, f"Turnover {turnover:.4f} above cap of {self.max_turnover}" # 5b. HTVR ratio check — log eligibility for x2/x3 multiplier campaign # HTVR ratio = returns / turnover (platform definition) returns_val = sim_results.get("returns") or sim_results.get("ann_ret") if returns_val is not None and turnover and turnover > 0: htvr_ratio = abs(returns_val) / turnover if turnover > 0.20 and htvr_ratio >= self.min_htvr_ratio: logger.info( f"HTVR ELIGIBLE: Turnover={turnover:.2%}, Returns={returns_val:.2%}, " f"HTVR_ratio={htvr_ratio:.2f} — qualifies for x2 High Turnover Multiplier" ) # 6. Parse and check checks list for Correlation and Sub-Universe NaN failures raw_data = sim_results.get("raw_data") or {} # Extract checks from raw JSON payload checks = [] alpha_data = raw_data.get("alpha", {}) if isinstance(alpha_data, dict): checks = alpha_data.get("is", {}).get("checks", []) if not checks and isinstance(raw_data, dict): checks = raw_data.get("is", {}).get("checks", []) or raw_data.get("checks", []) if not isinstance(checks, list): checks = [] self_corr = None prod_corr = None sub_universe_nan_fail = False for c in checks: if not isinstance(c, dict): continue name = str(c.get("name", "")).lower() val = c.get("value") passed = c.get("passed") result = str(c.get("result", "")).upper() msg = str(c.get("message", "")).lower() # Self-correlation check if "selfcorrelation" in name or "self_correlation" in name: try: self_corr = float(val) if val is not None else None except (ValueError, TypeError): pass if passed is False or result == "FAIL": return False, f"Self-correlation check failed: {msg or 'Exceeded platform limits'}" # Product-correlation check elif "prodcorrelation" in name or "prod_correlation" in name or "productioncorrelation" in name: try: prod_corr = float(val) if val is not None else None except (ValueError, TypeError): pass if passed is False or result == "FAIL": return False, f"Production-correlation check failed: {msg or 'Exceeded platform limits'}" # Sub-universe Sharpe NaN is not above cutoff check elif "sub-universe" in name or "subuniverse" in name or "sub-universe" in msg or "subuniverse" in msg: if passed is False or result == "FAIL" or "nan" in msg: sub_universe_nan_fail = True # General catch-all for any failed check in the simulation logs elif passed is False or result == "FAIL": return False, f"Platform check failed: {c.get('name')} - {msg}" # Apply custom correlation thresholds if values are parsed if self_corr is not None and self_corr >= self.max_self_correlation: return False, f"Self-correlation {self_corr:.4f} exceeds limit of {self.max_self_correlation}" if prod_corr is not None and prod_corr >= self.max_prod_correlation: return False, f"Production-correlation {prod_corr:.4f} exceeds limit of {self.max_prod_correlation}" if sub_universe_nan_fail: return False, "Sub-universe check failed: Sharpe is NaN or below cutoff in sub-universe" # 7. Strict Power Pool Theme Verification (only applies to Pure Power Pool candidates: Sharpe < 1.58) if config.ENABLE_PURE_POWER_POOL_SUBMISSIONS and config.STRICT_POWER_POOL_THEME and sharpe < 1.58: from datetime import datetime, date current_date = datetime.now().date() # Dataset restrictions: drop if contains pvl or pv1 expr_lower = expression.lower() if expression else "" if "pvl" in expr_lower or "pv1" in expr_lower: return False, "Failed theme check: Pure Power Pool candidate contains restricted dataset 'pvl' or 'pv1'." # Late July blocks 'model110' if current_date >= date(2026, 7, 27) and "model110" in expr_lower: return False, "Failed theme check: Pure Power Pool candidate contains restricted dataset 'model110'." # July 13 to July 26 requires High Turnover returns ratio >= 0.75 if date(2026, 7, 13) <= current_date <= date(2026, 7, 26): htvr_value = None for c in checks: if not isinstance(c, dict): continue c_name = str(c.get("name", "")).lower() if c_name == "ht_high_turnover_returns_ratio": try: htvr_value = float(c.get("value")) except (ValueError, TypeError): pass # Alternate calculation fallback (returns / turnover) if htvr_value is None: returns_val = sim_results.get("returns") or sim_results.get("ann_ret") if returns_val is not None and turnover and turnover > 0: htvr_value = abs(returns_val) / turnover if htvr_value is None or htvr_value < 0.75: return False, f"Failed theme check: Simulated HTVR ratio ({htvr_value if htvr_value is not None else 'N/A'}) is below 0.75 (required for Pure Power Pool July 13-26)." return True, "Passed all acceptance thresholds" async def evaluate_and_update(self, alpha_id: int, sim_results: Dict[str, Any], delay: int) -> str: """ Runs evaluation on the simulation results and updates the record in SQLite database. Stores categorized trash_reason for analytics observability. Returns the new status string. """ async with AsyncSessionLocal() as session: # Fetch the alpha record alpha = await session.get(Alpha, alpha_id) if not alpha: logger.error(f"Alpha record {alpha_id} not found in database for evaluation.") return "NOT_FOUND" expr = alpha.expression is_passing, reason = self.evaluate_metrics(sim_results, delay, expr) # Parse results and save metrics alpha.sharpe = sim_results.get("sharpe") alpha.fitness = sim_results.get("fitness") alpha.turnover = sim_results.get("turnover") alpha.raw_result_json = serialize_raw_payload(sim_results.get("raw_data")) if is_passing: alpha.status = "PENDING_SUBMIT" alpha.error_log = reason alpha.trash_reason = None logger.info(f"Alpha {alpha_id} PASSED evaluation. Queued for submission.") else: alpha.status = "TRASHED" alpha.error_log = reason alpha.trash_reason = classify_trash_reason(reason) logger.info(f"Alpha {alpha_id} TRASHED during evaluation: {reason}") await session.commit() return alpha.status