Alpha Autopilot

alpha-monitor → alpha-mine → alpha-evaluate → alpha-library → alpha-signal

import sqlite3, json, os
from datetime import datetime

PROJECT_DIR = "<current working directory>"
db_path = os.path.join(PROJECT_DIR, "alpha_skills.db")

# Read factor library
with sqlite3.connect(db_path) as conn:
    conn.row_factory = sqlite3.Row
    all_factors = conn.execute("SELECT * FROM factors ORDER BY status, icir DESC").fetchall()
    all_factors = [dict(r) for r in all_factors]

active = [f for f in all_factors if f["status"] == "active"]
warning = [f for f in all_factors if f["status"] == "warning"]
alert = [f for f in all_factors if f["status"] == "alert"]
retired = [f for f in all_factors if f["status"] == "retired"]

print(f"""
🤖 Autopilot Status / 自动驾驶状态
  Active 活跃: {len(active)}
  Warning 警告: {len(warning)}
  Alert 告警: {len(alert)}
  Retired 退役: {len(retired)}
""")

🤖 Autopilot Initiating / 自动驾驶启动

Phase 1: Health Check / 健康检查
  📚 Factor Library: {n} active, {n} warning, {n} alert

from scipy import stats

# Load recent data (same as alpha-evaluate data loading)
# Compute factor values and forward returns for last 60 trading days

def check_factor_health(factor_name, factor_values, forward_returns, registered_icir):
    """Check if a factor is still healthy"""
    # Compute recent IC
    recent_dates = factor_values.index[-60:]
    ic_values = []
    for date in recent_dates:
        f = factor_values.loc[date].dropna()
        r = forward_returns.loc[date].dropna() if date in forward_returns.index else pd.Series()
        common = f.index.intersection(r.index)
        if len(common) < 30:
            continue
        corr, _ = stats.spearmanr(f[common].values, r[common].values)
        if np.isfinite(corr):
            ic_values.append(corr)
    
    if len(ic_values) < 10:
        return "insufficient_data", 0, 0
    
    rolling_ic = np.mean(ic_values)
    rolling_icir = np.mean(ic_values) / np.std(ic_values) if np.std(ic_values) > 0 else 0
    
    # Compare with registered ICIR
    if registered_icir and registered_icir > 0:
        decay_ratio = rolling_icir / registered_icir
    else:
        decay_ratio = 1.0
    
    # Determine status
    if rolling_icir < 0:
        return "alert", rolling_icir, decay_ratio
    elif decay_ratio < 0.5:
        return "warning", rolling_icir, decay_ratio
    else:
        return "healthy", rolling_icir, decay_ratio

# Check each active factor
health_results = []
for f in active:
    name = f["name"]
    # Compute factor values using the expression/function
    # ... (same pattern as alpha-signal)
    status, rolling_icir, decay = check_factor_health(
        name, factor_vals, fwd_ret, f.get("icir", 0)
    )
    health_results.append({
        "name": name, "status": status, 
        "rolling_icir": rolling_icir, "decay": decay,
        "registered_icir": f.get("icir", 0)
    })

Phase 2: Factor Health Monitor / 因子健康监控

  🟢 pv_diverge       Rolling ICIR=0.62 (reg 0.70) Decay=11%  HEALTHY
  🟢 turnover_20      Rolling ICIR=0.48 (reg 0.52) Decay=8%   HEALTHY
  🟡 volatility_20    Rolling ICIR=0.19 (reg 0.43) Decay=56%  WARNING
  🔴 reversal_5       Rolling ICIR=-0.05 (reg 0.37) Decay=113% ALERT

# Factors with ALERT status for 2+ consecutive checks → auto-retire
for result in health_results:
    if result["status"] == "alert":
        name = result["name"]
        print(f"  🔴 Retiring {name}: ICIR turned negative")
        with sqlite3.connect(db_path) as conn:
            conn.execute("UPDATE factors SET status='retired' WHERE name=?", (name,))
    elif result["status"] == "warning":
        name = result["name"]
        print(f"  🟡 Downgrading {name} to WARNING status")
        with sqlite3.connect(db_path) as conn:
            conn.execute("UPDATE factors SET status='warning' WHERE name=?", (name,))

Phase 3: Lifecycle Actions / 生命周期操作

  🔴 reversal_5 → RETIRED (ICIR turned negative)
  🟡 volatility_20 → WARNING (ICIR decayed 56%)
  Need replacement: 1 factor retired

TARGET_LIBRARY_SIZE = 5  # aim for at least 5 active factors
current_active = len([r for r in health_results if r["status"] == "healthy"])
need_new = max(0, TARGET_LIBRARY_SIZE - current_active)

if need_new > 0:
    print(f"\n  Mining {need_new * 10} candidates to find {need_new} replacements...")
    # Run mining pipeline (same as alpha-mine)
    # Generate candidates → IC quick screen → full evaluate top candidates
    # ...

Phase 4: Factor Mining / 因子挖掘

  ⛏️ Generated 50 candidates
  📊 Passed IC screen: 8 (16%)
  🏆 Top discoveries:
    1. Low downside vol 20d    ICIR=0.53  Intuition: Strong
    2. Mean reversion MA40     ICIR=0.48  Intuition: Moderate
    3. Volume-weighted mom 10d ICIR=0.42  Intuition: Moderate

# Register candidates that pass quality threshold
QUALITY_THRESHOLD = "moderate"  # or "strong" for more conservative

for candidate in top_candidates:
    if candidate["quality"] in ["strong", "moderate"]:
        # Check correlation with existing factors
        # If correlation > 0.7 with any existing factor → skip (redundant)
        
        # Register
        with sqlite3.connect(db_path) as conn:
            import uuid
            fid = str(uuid.uuid4())[:8]
            conn.execute(
                "INSERT OR REPLACE INTO factors (id,name,expression,category,status,ic_mean,icir,quality,eval_date) VALUES (?,?,?,?,?,?,?,?,?)",
                (fid, candidate["name"], candidate["expr"], candidate["category"],
                 "active", candidate["ic_mean"], candidate["icir"], candidate["quality"],
                 datetime.now().isoformat())
            )
        print(f"  ✅ Registered: {candidate['name']} (ICIR={candidate['icir']:.3f})")

Phase 5: Auto-Register / 自动注册

  ✅ Registered: low_downside_vol_20d (ICIR=0.534, Strong)
  ⏭️ Skipped: mean_reversion_ma40 (corr=0.72 with volatility_20, redundant)

# Re-read updated library
# Run alpha-signal logic
# Output today's target portfolio

Phase 6: Signal Generation / 信号生成

  📡 Today's signal generated with {n} active factors
  Signal saved: signals/{date}.csv