NamVan56/QuantForge
0
1import asyncio2import logging3import time4import re5from database import init_db, SessionLocal, AlphaRecord6from wq_client import BrainClient7from agent import AlphaAgent, USED_LOCAL_FALLBACK8from submission_engine import SubmissionEngine9import config10import random11 12# Configure standard logger for orchestrator13logger = logging.getLogger("Orchestrator")14 15# Global Control Flags16MINER_RUNNING = False17MINER_TASK = None18GLOBAL_SUBMISSION_ENGINE = None19LAST_GENERATION_TIME = 0.020 21# Dynamic throttle: short for local fallback, longer for LLM22THROTTLE_LOCAL_FALLBACK = 10 # 10 seconds between local fallback generations23THROTTLE_LLM = 120 # 2 minutes between LLM-backed generations24 25def normalize_settings(settings: dict) -> dict:26 """27 Normalizes and validates settings to ensure compatibility with WorldQuant BRAIN API.28 """29 if not isinstance(settings, dict):30 settings = {}31 32 region = settings.get("region", "USA")33 if region == "US" or not region:34 region = "USA"35 else:36 region = str(region).upper()37 if region not in ["USA"]:38 region = "USA"39 40 universe = settings.get("universe", "TOP3000")41 if universe in ["univ_standard", "univ_top_90", "univ_top_2000"] or not universe:42 universe = "TOP3000"43 else:44 universe = str(universe).upper()45 if universe not in ["TOP3000", "TOP2000", "TOP1000", "TOP500", "TOP200", "TOPSP500"]:46 universe = "TOP3000"47 48 neutralization = settings.get("neutralization", "SUBINDUSTRY")49 if neutralization:50 neutralization = str(neutralization).upper()51 if neutralization not in ["SUBINDUSTRY", "INDUSTRY", "SECTOR", "MARKET", "NONE"]:52 neutralization = "SUBINDUSTRY"53 else:54 neutralization = "SUBINDUSTRY"55 56 try:57 delay = int(settings.get("delay", 1))58 if delay not in [0, 1]:59 delay = 160 except (ValueError, TypeError):61 delay = 162 63 64 try:65 decay = int(settings.get("decay", 10))66 if decay > 512:67 decay = 51268 elif decay < 0:69 decay = 070 except (ValueError, TypeError):71 decay = 1072 73 try:74 truncation = float(settings.get("truncation", 0.01))75 except (ValueError, TypeError):76 truncation = 0.0177 78 return {79 "region": region,80 "universe": universe,81 "neutralization": neutralization,82 "delay": delay,83 "decay": decay,84 "truncation": truncation85 }86 87VALID_OPERATORS = {88 # Mathematical89 "abs", "log", "sign", "sqrt", "exp", "power", "min", "max",90 # Cross-sectional91 "rank", "zscore", "group_rank", "group_neutralize", "group_zscore", "vector_neut",92 # Time-series93 "ts_delta", "ts_mean", "ts_std_dev", "ts_decay_linear", "signed_power", 94 "ts_skewness", "ts_kurt", "ts_corr", "ts_covariance", "ts_max", "ts_min", 95 "ts_rank", "ts_sum", "ts_product", "ts_argmax", "ts_argmin", "ts_scale",96 # Vector97 "vec_avg", "vec_sum", "vec_choose",98 # Constants/Groups99 "sector", "subindustry", "industry", "market", "cap",100 # Price-volume/basic fields101 "close", "open", "high", "low", "volume", "vwap", "returns",102 "adv20", "adv60", "sharesout", "adjfactor"103}104 105def sanitize_expression(expr: str) -> str:106 """107 Sanitizes WorldQuant expressions to ensure they are compatible with the platform.108 Converts scientific notation like '1e-9' or '1e-5' to full decimal notation (e.g. '0.000000001').109 """110 if not expr:111 return expr112 113 def replace_sci(match):114 sci_str = match.group(0)115 try:116 val = float(sci_str)117 dec_str = f"{val:.15f}".rstrip('0')118 if dec_str.endswith('.'):119 dec_str += '0'120 return dec_str121 except ValueError:122 return sci_str123 124 pattern = r'\b\d+(?:\.\d+)?[eE][+-]?\d+\b'125 return re.sub(pattern, replace_sci, expr)126 127def validate_expression(expr: str) -> bool:128 """129 Validates basic syntax rules and variable names of a generated alpha formula130 before submitting for simulation.131 """132 if not expr or len(expr.strip()) < 5:133 return False134 135 # Check balanced parentheses and brackets136 if expr.count('(') != expr.count(')'):137 return False138 if expr.count('[') != expr.count(']'):139 return False140 141 # Check for empty wrappers142 if "()" in expr or "[]" in expr:143 return False144 145 # Check for redundant nested wrappers146 if "rank(rank(" in expr or "zscore(zscore(" in expr:147 return False148 149 # Check for invalid variable names and vector variable operator wrapping if the cache exists150 import json151 cache_file = config.DATA_DIR / "data_fields_cache.json"152 if cache_file.exists():153 try:154 with open(cache_file, "r") as f:155 cached_data = json.load(f)156 if isinstance(cached_data, dict) and cached_data:157 # Gather all valid field IDs and vector fields from cache158 valid_fields = set()159 vector_fields = set()160 for cat_fields in cached_data.values():161 for f in cat_fields:162 fid = f.get("id")163 valid_fields.add(fid)164 if f.get("type") == "VECTOR":165 vector_fields.add(fid)166 167 # Extract all word tokens using regex168 tokens = re.findall(r'\b[a-zA-Z_][a-zA-Z0-9_]*\b', expr)169 for t in tokens:170 # Ignore numeric tokens/parameters and built-in operators/constants171 if t.lower() in VALID_OPERATORS or t in VALID_OPERATORS:172 continue173 # Ignore standard indices like sector, industry, etc.174 if t.lower() in ("sector", "subindustry", "industry", "market"):175 continue176 # Check if token is in valid cached fields177 if t not in valid_fields:178 logger.warning(f"Validation failed: Unknown variable '{t}' in formula '{expr}'")179 return False180 # Check if Vector field is wrapped in a vector operator (vec_avg, vec_sum, vec_choose)181 if t in vector_fields:182 pattern = r'\bvec_(avg|sum|choose)\s*\([^)]*\b' + re.escape(t) + r'\b'183 if not re.search(pattern, expr):184 logger.warning(f"Validation failed: Vector field '{t}' in formula '{expr}' is not wrapped in a vector operator.")185 return False186 except Exception as e:187 logger.warning(f"Error loading cache for variable validation: {e}")188 189 return True190 191def is_worth_refining(error_log: str, sharpe: float = None, fitness: float = None, turnover: float = None) -> bool:192 if not error_log:193 return False194 195 err_lower = error_log.lower()196 197 # 1. Syntax, unit, operator, variable, or platform check errors are worth refining198 syntax_keywords = [199 "unit", "variable", "operator", "syntax", "invalid input", 200 "must be an expression", "divide", "expected", "less than or equal to",201 "failed_checks", "failed checks", "concentrated_weight", "low_sub_universe_sharpe",202 "platform_check_failure", "concentrated", "sub-universe"203 ]204 if any(k in err_lower for k in syntax_keywords):205 # If it failed due to check failures, only refine if it has promising metrics206 if "check" in err_lower or "concentrated" in err_lower or "sub_universe" in err_lower or "sub-universe" in err_lower:207 if sharpe is not None and (sharpe < 0.8 or fitness < 0.3):208 return False209 return True210 211 # 2. Near-miss performance alphas (Tier 2 and above) are worth refining212 if sharpe is not None and fitness is not None and turnover is not None:213 is_near_miss = (sharpe >= 0.9 and fitness >= 0.4 and turnover >= 0.01 and turnover <= 0.70)214 if is_near_miss:215 return True216 217 return False218 219def clean_and_normalize_database():220 """221 Sweeps the entire database, normalizes all record settings,222 and resets any records that failed due to payload/settings format validation.223 """224 logger.info("Starting database settings normalization and status cleanup...")225 try:226 db = SessionLocal()227 records = db.query(AlphaRecord).all()228 229 # Phase 1: Group records by their target normalized expression230 groups = {}231 for record in records:232 expr = record.expression233 234 # Normalize invalid variable names to valid counterparts235 normalized = expr236 normalized = re.sub(r'\bmarket_cap(?:_usd)?\b', 'cap', normalized)237 normalized = re.sub(r'\bnet_income\b', 'ebit', normalized)238 normalized = re.sub(r'\bnetincome\b', 'ebit', normalized)239 normalized = re.sub(r'\bincome_q\b', 'ebit', normalized)240 normalized = re.sub(r'\bincome\b', 'ebit', normalized)241 normalized = re.sub(r'\bni\b', 'ebit', normalized)242 normalized = re.sub(r'\bsales\b', 'cashflow', normalized)243 normalized = re.sub(r'\beps_ttm\b', 'ebit', normalized)244 normalized = re.sub(r'\bearnings_per_share_ttm\b', 'ebit', normalized)245 normalized = re.sub(r'\bearnings_income\b', 'ebit', normalized)246 normalized = re.sub(r'\bearnings_ttm_diluted_eps\b', 'ebit', normalized)247 normalized = re.sub(r'\bcash_flow\b', 'cashflow', normalized)248 normalized = sanitize_expression(normalized)249 250 if normalized not in groups:251 groups[normalized] = []252 groups[normalized].append((record, normalized != expr))253 254 deleted_count = 0255 updated_count = 0256 257 records_to_keep = []258 records_to_delete = []259 260 # Phase 2: For each group, choose the best record to keep and delete others261 for normalized_expr, group in groups.items():262 if len(group) == 1:263 records_to_keep.append(group[0])264 else:265 # Group has multiple records. Rank them.266 status_ranks = {267 "SUBMITTED": 5,268 "PASSED_CORRELATION": 4,269 "PENDING_CORRELATION": 3,270 "GENERATED": 2,271 "SIMULATING": 1,272 "FAILED": 0273 }274 275 def sort_key(item):276 rec, is_normalized_different = item277 is_already_normalized = not is_normalized_different278 status_rank = status_ranks.get(rec.status, 0)279 return (is_already_normalized, status_rank, -rec.id)280 281 sorted_group = sorted(group, key=sort_key, reverse=True)282 283 # Keep the first, delete the rest284 records_to_keep.append(sorted_group[0])285 for rec, _ in sorted_group[1:]:286 records_to_delete.append(rec)287 288 # Phase 3: Delete duplicates first and commit to free up the unique constraint289 if records_to_delete:290 logger.info(f"Identifying {len(records_to_delete)} duplicate/conflicting records to delete...")291 for rec in records_to_delete:292 logger.debug(f"Deleting duplicate record: {rec.expression}")293 db.delete(rec)294 db.commit()295 deleted_count = len(records_to_delete)296 297 # Phase 4: Normalize the kept records and update them in the database298 existing_exprs = {rec.expression for rec, _ in records_to_keep}299 for rec, is_normalized_different in records_to_keep:300 mutated = False301 302 if is_normalized_different:303 normalized = rec.expression304 normalized = re.sub(r'\bmarket_cap(?:_usd)?\b', 'cap', normalized)305 normalized = re.sub(r'\bnet_income\b', 'ebit', normalized)306 normalized = re.sub(r'\bnetincome\b', 'ebit', normalized)307 normalized = re.sub(r'\bincome_q\b', 'ebit', normalized)308 normalized = re.sub(r'\bincome\b', 'ebit', normalized)309 normalized = re.sub(r'\bni\b', 'ebit', normalized)310 normalized = re.sub(r'\bsales\b', 'cashflow', normalized)311 normalized = re.sub(r'\beps_ttm\b', 'ebit', normalized)312 normalized = re.sub(r'\bearnings_per_share_ttm\b', 'ebit', normalized)313 normalized = re.sub(r'\bearnings_income\b', 'ebit', normalized)314 normalized = re.sub(r'\bearnings_ttm_diluted_eps\b', 'ebit', normalized)315 normalized = re.sub(r'\bcash_flow\b', 'cashflow', normalized)316 normalized = sanitize_expression(normalized)317 318 logger.debug(f"Normalizing expression: {rec.expression} -> {normalized}")319 rec.expression = normalized320 rec.status = "GENERATED"321 rec.error_log = None322 mutated = True323 324 # Sanitize scientific notation for ALL records (even those without variable renames)325 if not is_normalized_different:326 sanitized = sanitize_expression(rec.expression)327 if sanitized != rec.expression:328 logger.debug(f"Sanitizing sci notation: {rec.expression} -> {sanitized}")329 rec.expression = sanitized330 # Reset to GENERATED so it gets re-simulated with the fixed expression331 if rec.status in ("FAILED", "SIMULATING"):332 rec.status = "GENERATED"333 rec.error_log = None334 mutated = True335 336 # Normalize region337 normalized_region = "USA"338 if rec.region != normalized_region:339 rec.region = normalized_region340 mutated = True341 342 # Normalize universe343 normalized_universe = "TOP3000"344 if rec.universe:345 u_val = rec.universe.upper()346 if u_val in ["TOP3000", "TOP2000", "TOP1000", "TOP500", "TOP200", "TOPSP500"]:347 normalized_universe = u_val348 if rec.universe != normalized_universe:349 rec.universe = normalized_universe350 mutated = True351 352 # Normalize neutralization353 normalized_neut = "SUBINDUSTRY"354 if rec.neutralization:355 normalized_neut = rec.neutralization.upper()356 if normalized_neut not in ["SUBINDUSTRY", "INDUSTRY", "SECTOR", "MARKET", "NONE"]:357 normalized_neut = "SUBINDUSTRY"358 if rec.neutralization != normalized_neut:359 rec.neutralization = normalized_neut360 mutated = True361 362 # Normalize delay363 normalized_delay = 1364 try:365 if rec.delay is not None:366 d_val = int(rec.delay)367 if d_val in [0, 1]:368 normalized_delay = d_val369 except (ValueError, TypeError):370 pass371 if rec.delay != normalized_delay:372 rec.delay = normalized_delay373 mutated = True374 375 376 # Reset SIMULATING or PASSED records back to GENERATED on startup377 if rec.status in ("SIMULATING", "PASSED"):378 rec.status = "GENERATED"379 mutated = True380 381 # Reset FAILED records caused by payload/settings formatting or sci notation issues382 if rec.status == "FAILED" and rec.error_log:383 err_lower = rec.error_log.lower()384 if any(x in err_lower for x in ["settings", "choice", "is not a valid", "field is required", "submit failed: 400", "400 -", "unexpected character"]):385 rec.status = "GENERATED"386 rec.error_log = None387 mutated = True388 389 # Reset/demote any correlation-pending/passed records that don't meet strict requirements390 if rec.status in ("PENDING_CORRELATION", "PASSED_CORRELATION", "QUEUED_FOR_NEXT_BATCH"):391 if rec.sharpe is not None and rec.fitness is not None:392 if rec.sharpe < 1.25 or rec.fitness < 1.0:393 logger.info(f"Demoting sub-threshold alpha {rec.expression} (Sharpe={rec.sharpe:.3f}, Fitness={rec.fitness:.3f}) from status {rec.status} to FAILED.")394 rec.status = "FAILED"395 rec.error_log = f"Demoted: Sharpe={rec.sharpe:.3f} < 1.25 or Fitness={rec.fitness:.3f} < 1.0"396 mutated = True397 398 # Retroactively promote FAILED records that meet the strict criteria399 if rec.status == "FAILED" and rec.sharpe is not None and rec.fitness is not None and rec.turnover is not None:400 sharpe_ok = rec.sharpe >= 1.25401 fitness_ok = rec.fitness >= 1.0402 turnover_ok = 0.01 <= rec.turnover <= 0.70403 if sharpe_ok and fitness_ok and turnover_ok:404 logger.info(f"Retroactively promoting high-performing failed alpha {rec.expression} (Sharpe={rec.sharpe}, Fitness={rec.fitness}) back to GENERATED.")405 rec.status = "GENERATED"406 rec.error_log = None407 mutated = True408 elif abs(rec.sharpe) >= 1.25 and abs(rec.fitness) >= 1.0 and turnover_ok and rec.sharpe < 0:409 orig_expr = rec.expression.strip()410 if orig_expr.startswith("-"):411 if orig_expr.startswith("-("):412 flipped_expr = orig_expr[2:-1]413 else:414 flipped_expr = orig_expr[1:]415 else:416 flipped_expr = f"-({orig_expr})"417 418 if flipped_expr not in existing_exprs:419 logger.info(f"Discovered negative Sharpe alpha with high absolute metrics: {rec.expression} (Sharpe={rec.sharpe}). Adding sign-flipped version to DB: {flipped_expr}")420 new_rec = AlphaRecord(421 expression=flipped_expr,422 region=rec.region,423 universe=rec.universe,424 delay=rec.delay,425 decay=rec.decay,426 truncation=rec.truncation,427 neutralization=rec.neutralization,428 status="GENERATED"429 )430 db.add(new_rec)431 existing_exprs.add(flipped_expr)432 updated_count += 1433 434 if mutated:435 updated_count += 1436 437 if updated_count > 0 or deleted_count > 0:438 db.commit()439 logger.info(f"Successfully cleaned and normalized. Updated: {updated_count}, Deleted: {deleted_count} alpha records.")440 else:441 logger.info("Database is already clean and normalized.")442 443 db.close()444 except Exception as e:445 logger.error(f"Error during database normalization: {e}")446 447async def miner_loop():448 """449 Central execution loop running the autonomous Alpha Mining pipeline.450 """451 global MINER_RUNNING, LAST_GENERATION_TIME452 logger.info("Initializing database...")453 init_db()454 455 # Run database normalization and startup cleanup456 clean_and_normalize_database()457 458 global GLOBAL_SUBMISSION_ENGINE459 460 client = BrainClient()461 agent = AlphaAgent()462 if GLOBAL_SUBMISSION_ENGINE is None:463 GLOBAL_SUBMISSION_ENGINE = SubmissionEngine(client)464 465 # Ensure data fields are cached in the background466 asyncio.create_task(client.fetch_and_cache_data_fields(["USA"], ["TOP3000", "TOP2000", "TOP1000", "TOP500", "TOP200", "TOPSP500"]))467 468 # Run startup synchronization to realign local DB status of submitted alphas469 try:470 await GLOBAL_SUBMISSION_ENGINE.sync_submitted_alphas()471 except Exception as e:472 logger.error(f"Startup synchronization of submitted alphas failed: {e}")473 474 # Process/submit any legacy queued alphas before activating new loop475 try:476 await GLOBAL_SUBMISSION_ENGINE.clear_legacy_queue()477 except Exception as e:478 logger.error(f"Startup clearing of legacy submission queue failed: {e}")479 480 # Default fallback data fields481 data_fields = ["ebit", "capex", "assets", "debt", "equity", "cash", "cashflow"]482 483 while MINER_RUNNING:484 try:485 db = SessionLocal()486 487 # ==========================================488 # PHASE A: Generation489 # ==========================================490 # We first check if we already have sufficient pending/unprocessed alphas491 pending_count = db.query(AlphaRecord).filter(492 AlphaRecord.status.in_(["GENERATED", "REFINED"])493 ).count()494 495 if pending_count < 150:496 now = time.time()497 # Dynamic throttle based on last generation source498 throttle_seconds = THROTTLE_LOCAL_FALLBACK if USED_LOCAL_FALLBACK else THROTTLE_LLM499 500 if now - LAST_GENERATION_TIME >= throttle_seconds:501 logger.info(f"Phase A: Queue size {pending_count} is low (target: 150). Generating new alphas...")502 503 # Rich library of proven seed templates to rotate for diversity and quality504 PROVEN_SEEDS = [505 "-ts_decay_linear(rank(ts_delta(rank(ebit), 10)), 45) * ts_rank(returns, 120)",506 "-ts_decay_linear(rank(ts_delta(rank(capex), 5)), 45) * ts_rank(returns, 252)",507 "-ts_decay_linear(rank(ts_delta(rank(assets), 5)), 40) * ts_rank(returns, 252)",508 "-ts_decay_linear(rank(ts_delta(rank(sales), 5)), 40) * ts_rank(returns, 252)",509 "-ts_decay_linear(rank(ts_delta(rank(ebit / cashflow), 5)), 60) * ts_rank(returns, 252)",510 "-rank(ts_decay_linear(rank(cashflow / (debt + 0.000000001)), 20)) * ts_rank(returns, 20)",511 "-ts_decay_linear(rank(ts_delta(rank(equity), 5)), 45) * ts_rank(returns, 252)",512 "-ts_decay_linear(rank(ts_delta(rank(cashflow), 5)), 40) * ts_rank(returns, 252)",513 "-ts_decay_linear(rank(cashflow / assets), 20) * ts_rank(returns, 20)",514 "-ts_decay_linear(rank(rank(cashflow) / rank(debt)), 40) * ts_rank(returns, 20)",515 "group_neutralize(rank(ebit / assets) - rank(cashflow / debt), SUBINDUSTRY)",516 "-ts_decay_linear(rank(ts_corr(close, volume, 20)) * rank(ts_std_dev(returns, 20)), 120)"517 ]518 519 seed_expression = random.choice(PROVEN_SEEDS)520 try:521 successful_alphas = db.query(AlphaRecord).filter(522 AlphaRecord.status.in_(["PASSED_CORRELATION", "SUBMITTED", "REFERENCE"])523 ).order_by(AlphaRecord.sharpe.desc()).all()524 525 if successful_alphas:526 # 70% chance of using a historical successful alpha, 30% chance of using a proven seed to maintain diversity527 if random.random() < 0.7:528 best_alpha = successful_alphas[0]529 seed_expression = best_alpha.expression530 logger.info(f"Selected best performer as dynamic seed alpha: {seed_expression} (Sharpe: {best_alpha.sharpe})")531 else:532 logger.info(f"Selected proven seed alpha from library for diversity: {seed_expression}")533 else:534 logger.info(f"No successful alphas in DB. Selected proven seed alpha: {seed_expression}")535 except Exception as e:536 logger.warning(f"Error querying dynamic seed alpha: {e}")537 538 # Query up to 5 successful example expressions to guide mutations539 try:540 successful_records = db.query(AlphaRecord).filter(541 AlphaRecord.status.in_(["PASSED_CORRELATION", "SUBMITTED", "REFERENCE"])542 ).order_by(AlphaRecord.sharpe.desc()).limit(5).all()543 successful_examples = [r.expression for r in successful_records]544 if successful_examples:545 logger.info(f"Using {len(successful_examples)} historical successful alphas as few-shot examples.")546 except Exception as e:547 logger.warning(f"Failed to query successful alphas for examples: {e}")548 successful_examples = []549 550 mutations = await agent.generate_mutations(seed_expression, data_fields, successful_examples=successful_examples)551 552 added_count = 0553 for mut in mutations:554 expression = mut.get("expression")555 raw_settings = mut.get("settings", {})556 if not expression:557 continue558 expression = sanitize_expression(expression)559 if not validate_expression(expression):560 continue561 existing = db.query(AlphaRecord).filter_by(expression=expression).first()562 if not existing:563 norm_settings = normalize_settings(raw_settings)564 new_record = AlphaRecord(565 expression=expression,566 region=norm_settings["region"],567 universe=norm_settings["universe"],568 delay=norm_settings["delay"],569 decay=norm_settings["decay"],570 truncation=norm_settings["truncation"],571 neutralization=norm_settings["neutralization"],572 status="GENERATED"573 )574 db.add(new_record)575 added_count += 1576 db.commit()577 LAST_GENERATION_TIME = time.time()578 579 pending_count = db.query(AlphaRecord).filter(580 AlphaRecord.status.in_(["GENERATED", "REFINED"])581 ).count()582 logger.info(f"Phase A complete. Added {added_count} alphas. Current pending: {pending_count}")583 else:584 time_remaining = int(throttle_seconds - (now - LAST_GENERATION_TIME))585 logger.info(f"Phase A: Queue low ({pending_count}), generation throttled. Re-evaluating in {time_remaining}s...")586 else:587 logger.info(f"Phase A: Skipped generation. {pending_count} alphas currently in queue.")588 589 # ==========================================590 # PHASE B: Simulation591 # ==========================================592 logger.info("Phase B: Querying alphas for simulation...")593 batch = db.query(AlphaRecord).filter(594 AlphaRecord.status.in_(["GENERATED", "REFINED"]),595 AlphaRecord.retry_count <= 5596 ).order_by(AlphaRecord.retry_count.asc(), AlphaRecord.id.desc()).limit(config.get_concurrent_limit()).all()597 598 if not batch:599 logger.info("No alphas to simulate. Sleeping for 10 seconds.")600 db.close()601 await asyncio.sleep(10)602 continue603 604 tasks = []605 for record in batch:606 record.status = "SIMULATING"607 norm_settings = normalize_settings({608 "region": record.region,609 "universe": record.universe,610 "delay": record.delay,611 "decay": record.decay,612 "truncation": record.truncation,613 "neutralization": record.neutralization614 })615 alpha_params = {616 "type": "REGULAR",617 "regular": record.expression,618 "settings": {619 "instrumentType": "EQUITY",620 "region": norm_settings["region"],621 "universe": norm_settings["universe"],622 "delay": norm_settings["delay"],623 "decay": norm_settings["decay"],624 "truncation": norm_settings["truncation"],625 "neutralization": norm_settings["neutralization"],626 "pasteurization": "ON",627 "unitHandling": "VERIFY",628 "nanHandling": "OFF",629 "language": "FASTEXPR",630 "visualization": False631 }632 }633 # Stagger submissions by len(tasks) * 4.0 seconds to avoid thundering herd and 429 collisions634 async def run_staggered(params, delay):635 if delay > 0:636 await asyncio.sleep(delay)637 return await client.simulate_alpha_async(params)638 639 tasks.append(run_staggered(alpha_params, len(tasks) * 4.0))640 641 db.commit() # Save 'SIMULATING' status642 643 logger.info(f"Submitting {len(tasks)} alphas for simulation concurrently (staggered)...")644 results_list = await asyncio.gather(*tasks, return_exceptions=True)645 646 # ==========================================647 # PHASE C: Evaluation & Refinement648 # ==========================================649 logger.info("Phase C: Evaluating simulation results...")650 651 for i, result in enumerate(results_list):652 record = batch[i]653 db.refresh(record)654 655 # Retrieve metrics if they were successfully simulated and saved in DB656 sharpe = record.sharpe657 fitness = record.fitness658 turnover = record.turnover659 660 if isinstance(result, Exception):661 logger.error(f"Simulation task for {record.expression} raised an exception: {result}")662 record.status = "FAILED"663 record.error_log = str(result)664 db.commit()665 continue666 667 if isinstance(result, dict) and result.get("status") == "FAILED" and record.status == "SIMULATING":668 record.status = "FAILED"669 record.error_log = result.get("error_message")670 db.commit()671 continue672 673 # AIR-TIGHT PERFORMANCE EVALUATION (Evaluates metrics if they exist, regardless of status)674 if sharpe is not None and fitness is not None and turnover is not None:675 turnover_max_ok = turnover <= 0.70676 turnover_min_ok = turnover >= 0.01677 turnover_ok = turnover_max_ok and turnover_min_ok678 679 tier1_ok = (sharpe >= 1.25) and (fitness >= 1.0) and turnover_ok680 tier2_ok = (sharpe >= 0.9) and (fitness >= 0.4) and turnover_ok681 682 wq_submittable = result.get("is_submittable") if isinstance(result, dict) else None683 684 # Check for platform check failures685 has_check_failures = False686 is_correlation_failure = False687 if record.error_log:688 err_lower = record.error_log.lower()689 has_check_failures = "failed" in err_lower or "fail" in err_lower690 is_correlation_failure = "correlation" in err_lower or "overlap" in err_lower or "similar" in err_lower691 692 if (tier1_ok or wq_submittable) and not has_check_failures:693 submit_reason = "WQ API is_submittable=True" if wq_submittable and not tier1_ok else "local criteria"694 logger.info(695 f"✅ SUCCESS (Tier 1)! Alpha {record.expression} passed ({submit_reason}): "696 f"Sharpe={sharpe:.3f}, Fitness={fitness:.3f}, Turnover={turnover:.3f}"697 )698 record.status = "PENDING_CORRELATION"699 record.error_log = None700 db.commit()701 elif (tier2_ok or tier1_ok) and is_correlation_failure:702 logger.info(703 f"⚠️ Tier 2 or Correlation Failure! Alpha {record.expression}: "704 f"Sharpe={sharpe:.3f}, Fitness={fitness:.3f}, Turnover={turnover:.3f}. "705 f"Queueing for orthogonalization."706 )707 record.status = "QUEUED_FOR_ORTHOGONALIZATION"708 record.error_log = f"TIER2_PASS: Sharpe={sharpe:.3f}, Fitness={fitness:.3f}, Turnover={turnover:.3f}"709 db.commit()710 elif (tier2_ok or tier1_ok) and has_check_failures:711 # Good performance but has a platform check failure (e.g. CONCENTRATED_WEIGHT) -> Refine to fix it!712 error_reason = f"PLATFORM_CHECK_FAILURE: Sharpe={sharpe:.3f}, Fitness={fitness:.3f}, failed: {record.error_log}"713 logger.info(f"⚠️ Promising alpha {record.expression} failed platform checks: {error_reason}")714 record.status = "FAILED"715 record.error_log = error_reason716 db.commit()717 elif tier2_ok:718 logger.info(719 f"⚠️ Tier 2 Pass! Alpha {record.expression}: "720 f"Sharpe={sharpe:.3f}, Fitness={fitness:.3f}, Turnover={turnover:.3f}. "721 f"Queueing for orthogonalization."722 )723 record.status = "QUEUED_FOR_ORTHOGONALIZATION"724 record.error_log = f"TIER2_PASS: Sharpe={sharpe:.3f}, Fitness={fitness:.3f}, Turnover={turnover:.3f}"725 db.commit()726 else:727 reasons = []728 if not (sharpe >= 1.25):729 reasons.append(f"Sharpe={sharpe:.3f} < 1.25")730 if not (fitness >= 1.0):731 reasons.append(f"Fitness={fitness:.3f} < 1.0")732 if not turnover_ok:733 reasons.append(f"Turnover={turnover:.3f} outside [1%, 70%]")734 735 is_near_miss = (sharpe >= 1.0 and fitness >= 0.7 and turnover_ok)736 737 if is_near_miss:738 error_reason = f"NEAR_MISS: Close but failed criteria: {', '.join(reasons)}"739 logger.info(f"⚠️ Near-miss alpha {record.expression}: {error_reason}")740 record.status = "FAILED"741 record.error_log = error_reason742 db.commit()743 else:744 error_reason = f"Failed criteria: {', '.join(reasons)}"745 logger.info(f"Alpha {record.expression} completed but {error_reason} (not worth refining)")746 record.status = "FAILED_REJECTED"747 record.error_log = error_reason748 db.commit()749 else:750 # No metrics returned, failed to simulate or compile751 if record.status == "SIMULATING":752 record.status = "FAILED"753 record.error_log = "Unknown compilation or simulation failure."754 db.commit()755 756 db.close()757 758 # ==========================================759 # PHASE D: Submission & Correlation Check760 # ==========================================761 logger.info("Phase D: Submission & Correlation Check...")762 await GLOBAL_SUBMISSION_ENGINE.process_pending_correlations()763 764 # ==========================================765 # PHASE E: Orthogonalization766 # ==========================================767 logger.info("Phase E: Processing alphas queued for orthogonalization...")768 db = SessionLocal()769 try:770 ortho_candidates = db.query(AlphaRecord).filter_by(771 status="QUEUED_FOR_ORTHOGONALIZATION"772 ).limit(5).all() # Process up to 5 at a time to prevent rate limits773 774 for record in ortho_candidates:775 logger.info(f"Orthogonalizing alpha: {record.expression} (correlation failure: {record.error_log})")776 777 alpha_dict = {778 "expression": record.expression,779 "settings": {780 "region": record.region,781 "universe": record.universe,782 "delay": record.delay,783 "decay": record.decay,784 "truncation": record.truncation,785 "neutralization": record.neutralization786 }787 }788 789 refined_alpha = await agent.orthogonalize_alpha(alpha_dict, record.error_log or "Correlation failure")790 if refined_alpha:791 new_expr = refined_alpha.get("expression")792 raw_settings = refined_alpha.get("settings", {})793 794 if new_expr and new_expr != record.expression:795 new_expr = sanitize_expression(new_expr)796 if not validate_expression(new_expr):797 logger.info(f"Skipping orthogonalized formula with syntax/variable errors: {new_expr}")798 record.status = "FAILED_CORRELATION" # Fail original and don't retry799 continue800 801 existing_refined = db.query(AlphaRecord).filter_by(expression=new_expr).first()802 if not existing_refined:803 norm_settings = normalize_settings(raw_settings)804 new_record = AlphaRecord(805 expression=new_expr,806 region=norm_settings["region"],807 universe=norm_settings["universe"],808 delay=norm_settings["delay"],809 decay=norm_settings["decay"],810 truncation=norm_settings["truncation"],811 neutralization=norm_settings["neutralization"],812 status="GENERATED"813 )814 db.add(new_record)815 logger.info(f"Successfully added orthogonalized alpha to database: {new_expr}")816 817 # Mark original alpha as FAILED_CORRELATION so we don't orthogonalize it again818 record.status = "FAILED_CORRELATION"819 db.commit()820 except Exception as e:821 logger.error(f"Error during Phase E Orthogonalization: {e}")822 db.rollback()823 finally:824 db.close()825 826 # ==========================================827 # PHASE F: Refinement of Failed Alphas828 # ==========================================829 logger.info("Phase F: Processing failed alphas for refinement...")830 db = SessionLocal()831 try:832 failed_candidates = db.query(AlphaRecord).filter_by(833 status="FAILED"834 ).limit(5).all() # Process up to 5 at a time835 836 for record in failed_candidates:837 # Differentiate between rate limit errors and other failures838 if record.error_log and any(x in record.error_log for x in ["CONCURRENT_SIMULATION_LIMIT_EXCEEDED", "rate limit", "Too Many Requests"]):839 record.retry_count = (record.retry_count or 0) + 1840 logger.info(f"Alpha {record.expression} failed due to rate limits/transient error (retry count: {record.retry_count}). Resetting to GENERATED.")841 record.status = "GENERATED"842 record.error_log = None843 db.commit()844 continue845 846 if not is_worth_refining(record.error_log, record.sharpe, record.fitness, record.turnover):847 logger.info(f"Alpha {record.expression} is not worth refining. Demoting to FAILED_REJECTED.")848 record.status = "FAILED_REJECTED"849 db.commit()850 continue851 852 logger.info(f"Alpha {record.expression} is worth refining. Triggering LLM self-correction...")853 alpha_dict = {854 "expression": record.expression,855 "settings": {856 "region": record.region,857 "universe": record.universe,858 "delay": record.delay,859 "decay": record.decay,860 "truncation": record.truncation,861 "neutralization": record.neutralization862 }863 }864 refined_alpha = await agent.refine_error(alpha_dict, record.error_log or "Unknown Error")865 if refined_alpha:866 new_expr = refined_alpha.get("expression")867 raw_settings = refined_alpha.get("settings", {})868 869 if new_expr and new_expr != record.expression:870 new_expr = sanitize_expression(new_expr)871 if not validate_expression(new_expr):872 logger.info(f"Skipping refined formula with syntax/variable errors: {new_expr}")873 record.status = "FAILED_REJECTED"874 db.commit()875 continue876 877 existing_refined = db.query(AlphaRecord).filter_by(expression=new_expr).first()878 if not existing_refined:879 norm_settings = normalize_settings(raw_settings)880 new_record = AlphaRecord(881 expression=new_expr,882 region=norm_settings["region"],883 universe=norm_settings["universe"],884 delay=norm_settings["delay"],885 decay=norm_settings["decay"],886 truncation=norm_settings["truncation"],887 neutralization=norm_settings["neutralization"],888 status="REFINED"889 )890 db.add(new_record)891 logger.info(f"Added successfully refined alpha to database: {new_expr}")892 893 # Mark the original record as FAILED_REJECTED so we don't refine it again894 record.status = "FAILED_REJECTED"895 db.commit()896 except Exception as e:897 logger.error(f"Error during Phase F Refinement: {e}")898 db.rollback()899 finally:900 db.close()901 902 # Short sleep before the next overarching iteration903 logger.info("Cycle complete. Sleeping before next iteration...")904 await asyncio.sleep(5)905 906 except Exception as e:907 logger.exception("Critical error in miner_loop. Sleeping before retry...")908 await asyncio.sleep(10)909 910if __name__ == "__main__":911 try:912 MINER_RUNNING = True913 asyncio.run(miner_loop())914 except KeyboardInterrupt:915 MINER_RUNNING = False916 logger.info("Miner gracefully shut down by user.")917