alpha-monitor

alpha-monitor — Factor Health Monitoring / 因子健康监控

你是一个因子监控系统。检查因子库中所有活跃因子的当前健康状态。 You are a factor monitoring system. Check health status of all active factors in the library.

Bilingual Terms / 双语术语

English	中文
Factor	因子
IC (Information Coefficient)	信息系数
ICIR (IC Information Ratio)	IC信息比率
Quintile	五分位/分组
Long-Short	多空
Sharpe Ratio	夏普比率
Max Drawdown	最大回撤
Monotonicity	单调性
Robustness	鲁棒性
Holding Period	持有期
Factor Registry	因子注册表
Backtest	回测
Gate Check	门控检查

项目定位 / Project Context

Multi-Market Support / 多市场支持:

Alpha Skills support A-share (default), HK, and US stocks via data adapters: Alpha Skills 通过数据适配器支持A股（默认）、港股和美股：

# .claude/alpha-agent.config.md
MARKET: A-share           # or "HK" or "US"
DATA_MODULE: (leave empty for A-share Tushare default)
                          # or "examples.us_data_yfinance"
                          # or "examples.hk_data_yfinance"

When a custom DATA_MODULE is set, the skill loads MARKET_CONFIG from that module to determine benchmark, cost rate, and trading rules. 设置自定义DATA_MODULE时，skill从该模块加载MARKET_CONFIG来确定基准、成本和交易规则。

Language Rule / 语言规则:

• If the user speaks English, output in English
• If the user speaks Chinese, output in Chinese
• Table headers always show both languages: "IC Mean IC均值"

执行流程 / Execution Pipeline

Step 1: 读取因子库 / Read Factor Library

import sys, os, sqlite3, json, uuid
from datetime import datetime

PROJECT_DIR = '<当前工作目录 current working directory>'

# ── 因子注册表（自包含，无外部依赖）/ Factor Registry (self-contained, no external deps) ──

class FactorRegistry:
    def __init__(self, db_path="alpha_skills.db"):
        self.db_path = db_path
        with sqlite3.connect(db_path) as conn:
            conn.execute("""CREATE TABLE IF NOT EXISTS factors (
                id TEXT PRIMARY KEY, name TEXT UNIQUE NOT NULL,
                expression TEXT NOT NULL, category TEXT, description TEXT,
                status TEXT DEFAULT 'active', market TEXT DEFAULT 'A-share',
                ic_mean REAL, icir REAL, best_holding_period INTEGER,
                quality TEXT, eval_date TEXT,
                created_at TEXT DEFAULT CURRENT_TIMESTAMP, metadata TEXT)""")
    
    def register(self, name, expression, **kwargs):
        fid = str(uuid.uuid4())[:8]
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT OR REPLACE INTO factors (id,name,expression,category,description,status,market,ic_mean,icir,best_holding_period,quality,eval_date,metadata) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
                (fid, name, expression, kwargs.get("category",""), kwargs.get("description",""),
                 "active", kwargs.get("market","A-share"), kwargs.get("ic_mean"), kwargs.get("icir"),
                 kwargs.get("best_holding_period"), kwargs.get("quality"),
                 datetime.now().isoformat(), json.dumps(kwargs.get("metadata",{}))))
        return fid
    
    def list_all(self, status=None):
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            if status:
                rows = conn.execute("SELECT * FROM factors WHERE status=? ORDER BY icir DESC", (status,)).fetchall()
            else:
                rows = conn.execute("SELECT * FROM factors ORDER BY icir DESC").fetchall()
            return [dict(r) for r in rows]
    
    def get(self, name):
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            row = conn.execute("SELECT * FROM factors WHERE name=?", (name,)).fetchone()
            return dict(row) if row else None
    
    def update_status(self, name, status):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("UPDATE factors SET status=? WHERE name=?", (status, name))
    
    def delete(self, name):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("DELETE FROM factors WHERE name=?", (name,))

reg = FactorRegistry(os.path.join(PROJECT_DIR, "alpha_skills.db"))
active_factors = reg.list_all(status='active')