Wq Alpha Research

WQ Alpha 研究 Skill

结构化 playbook：字段 → 表达式 → 回测 → 检查 → 提交 → 组合。融合 WorldQuant BRAIN 文档知识与 USA TOP3000 实证经验。

1. 快速决策树

开始
  ├── 拉取所有 alpha 列表 ──→ 只看 ACTIVE；算 **日收益** 相关，>0.7 则修改或放弃
  ├── 设计新因子
  │    ├── 字段已验证？ ──否──→ 查第 2 节（本地字段文件搜索 / 模拟 rank(field)）
  │    └── 是
  │         ├── 基本面 ──→ group_rank + ts_rank, SUBINDUSTRY, decay=0
  │         ├── 分析师 ──→ group_rank + ts_rank, INDUSTRY/SUBINDUSTRY, decay=0–4
  │         ├── 技术 ────→ 高 decay(10–30) 或混合基本面降低换手
  │         └── 情绪 ────→ nanHandling=ON, 小窗口谨慎
  └── 提交后 ──→ 验证 status == ACTIVE，否则检查 SELF_CORRELATION

2. 字段速查（本地数据集）

本 SKILL 已内置 USA TOP3000 delay=1 的完整字段列表（共 4367 个），无需每次从网页/ API 拉取：

• references/wq_usa_top3000_delay1_data_fields.json：完整字段元数据数组
• references/wq_usa_top3000_delay1_data_fields.csv：CSV 版，方便 Excel/ pandas 查看
• references/wq_usa_top3000_delay1_data_fields_summary.json：分类统计与示例字段

字段分布：

2.1 本地搜索字段

import json
from pathlib import Path

# 假设在 skill 目录下运行；如在其他位置，改为实际路径
skill_dir = Path(".")
field_dir = skill_dir / "references"
data = json.loads((field_dir / "wq_usa_top3000_delay1_data_fields.json").read_text(encoding="utf-8"))

keyword = "operating_income"
matches = [
    f for f in data
    if keyword.lower() in f["id"].lower()
    or (f.get("description") and keyword.lower() in f["description"].lower())
]

for f in matches[:10]:
    print(f"{f['id']} | {f.get('category',{}).get('name')} | {f.get('dataset',{}).get('name')} | coverage={f.get('coverage')} | alphaCount={f.get('alphaCount')}")

2.2 按类别筛选

category = "pv"  # 或 fundamental / analyst / news / option / model / socialmedia
fields = [f for f in data if f.get("category", {}).get("id") == category]
print(f"{category}: {len(fields)} fields")
for f in sorted(fields, key=lambda x: x.get("alphaCount", 0), reverse=True)[:10]:
    print(f"  {f['id']} | alphaCount={f.get('alphaCount')} | coverage={f.get('coverage')}")

2.3 字段验证

拿到候选字段后，先用简单表达式模拟验证字段是否真的可用：

payload = {
    "type": "REGULAR",
    "settings": {
        "instrumentType": "EQUITY", "region": "USA", "universe": "TOP3000",
        "delay": 1, "decay": 0, "neutralization": "MARKET",
        "truncation": 0.08, "pasteurization": "ON", "unitHandling": "VERIFY",
        "nanHandling": "ON", "language": "FASTEXPR", "visualization": False,
    },
    "regular": "rank(my_candidate_field)",
}
resp = session.post("https://api.worldquantbrain.com/simulations", json=payload)
# 201 表示字段可用；非 201 通常表示字段不存在或参数不匹配

2.4 何时需要重新拉取

本地字段集已覆盖 USA TOP3000 delay=1。以下情况才需要重新从 BRAIN 拉取：

• 换 Region（如 CHN、EUR）
• 换 Universe（如 TOP500、TOP1000）
• 换 Delay（如 0）
• BRAIN 平台字段列表明显更新（可对比 dateCreated 与本地）

3. 运算符速查表

黄金组合：group_rank(ts_rank(signal, N), subindustry)

4. 因子模板库

4.1 高胜率模板

-- 模板 A：ROE 趋势（通过率最高）
group_rank(ts_rank(operating_income / equity, 126), subindustry)

-- 模板 B：EPS 收益率修正
group_rank(ts_rank(est_eps / close, 126), industry)

-- 模板 C：FCF 收益率
group_rank(ts_rank(free_cash_flow_reported_value / equity, 126), industry)

-- 模板 D：多因子混合（高 Fitness）
0.5 * group_rank(ts_rank(operating_income / equity, 126), subindustry)
+ 0.5 * group_rank(ts_rank(est_eps / close, 126), industry)

-- 模板 E：低相关技术+基本面混合
0.5 * rank(-(close / open - 1)) + 0.5 * rank(ts_rank(operating_income / equity, 126))

-- 模板 F：资产周转 × 利润率
rank(ts_rank(operating_income / sales * sales / assets, 126))

4.2 推荐默认设置

5. 指标与检查

5.1 核心指标

5.2 IS 检查清单

5.3 失败统计

按数据类型通过率：基本面 40% > 混合 12.7% > 纯技术 5.3% > 其他 0%

6. 问题诊断与修复

7. BRAIN API 自动化

7.1 认证（请填写账号）

使用前必须准备凭据。推荐使用环境变量；也可以在本地放置未跟踪的 credential.txt（已被 .gitignore 忽略），内容为 JSON 数组：

["your_username", "your_password"]

⚠️ 提醒：不要把真实账号密码写入仓库。优先使用 WQ_BRAIN_USERNAME / WQ_BRAIN_PASSWORD 环境变量。

import json
import requests
from requests.auth import HTTPBasicAuth

API_BASE = "https://api.worldquantbrain.com"

# 1. 读取 credential.txt
import os

username = os.getenv("WQ_BRAIN_USERNAME")
password = os.getenv("WQ_BRAIN_PASSWORD")
if not (username and password):
    with open("credential.txt") as f:
        username, password = json.load(f)

# 2. 创建会话并认证
session = requests.Session()
session.auth = HTTPBasicAuth(username, password)
session.headers.update({
    "Content-Type": "application/json",
    "Accept": "application/json",
})

resp = session.post(f"{API_BASE}/authentication")
assert resp.status_code == 201, f"认证失败: {resp.status_code} {resp.text}"
print("认证成功")

7.2 获取已提交 Alpha 并计算相关性

目的：在新因子提交前，避免与已有因子 PnL 高度相关（相关系数 ≥ 0.7）。

import numpy as np

def fetch_pnl(session, alpha_id):
    """获取 Alpha 累计 PnL 序列；schema.properties 可能是 list 或 dict。"""
    r = session.get(f"{API_BASE}/alphas/{alpha_id}/recordsets/pnl")
    if r.status_code != 200 or not r.text.strip():
        return []
    data = r.json()
    props = data.get("schema", {}).get("properties", [])
    if isinstance(props, list):
        date_idx = next((i for i, p in enumerate(props) if p.get("name", "").lower() == "date"), 0)
        pnl_idx = next((i for i, p in enumerate(props) if p.get("name", "").lower() in ("pnl", "cum_pnl", "returns", "ret")), 1)
    else:
        date_idx = next((v["index"] for k, v in props.items() if k.lower() == "date"), 0)
        pnl_idx = next((v["index"] for k, v in props.items() if k.lower() in ("pnl", "cum_pnl", "returns", "ret")), 1)
    records = sorted(data.get("records", []), key=lambda r: r[date_idx])
    out = []
    for row in records:
        rec = row[0] if isinstance(row, list) and len(row) == 1 and isinstance(row[0], list) else row
        try:
            out.append(float(rec[pnl_idx]))
        except Exception:
            continue
    return out

def daily_returns(cum_pnl):
    """累计 PnL 转日收益；相关性应基于日收益，而非累计曲线。"""
    return [cum_pnl[i+1] - cum_pnl[i] for i in range(len(cum_pnl) - 1)]

def get_active_alphas(session, user_id="self", limit=100):
    """获取所有 alpha（含 ACTIVE / UNSUBMITTED），分页。"""
    all_alphas = []
    offset = 0
    while True:
        data = session.get(f"{API_BASE}/users/{user_id}/alphas", params={"limit": limit, "offset": offset}).json()
        batch = data.get("results", data.get("alphas", []))
        if not batch:
            break
        all_alphas.extend(batch)
        if len(batch) < limit:
            break
        offset += limit
    return all_alphas

# 计算新因子与所有 ACTIVE alpha 的日收益相关性
new_pnl = fetch_pnl(session, new_alpha_id)
new_ret = daily_returns(new_pnl)
existing = get_active_alphas(session)
active = [a for a in existing if a.get("status") == "ACTIVE"]

high_corr = []
for alpha in active:
    old_id = alpha.get("id")
    try:
        old_pnl = fetch_pnl(session, old_id)
        old_ret = daily_returns(old_pnl)
        if len(new_ret) == len(old_ret) and len(new_ret) > 20:
            corr = float(np.corrcoef(new_ret, old_ret)[0, 1])
            print(f"与 {old_id} 日收益相关性: {corr:.3f}")
            if abs(corr) >= 0.7:
                high_corr.append((old_id, corr))
    except Exception:
        continue

if high_corr:
    print(f"⚠️ 发现 {len(high_corr)} 个高相关因子，建议修改或放弃")

判断规则（基于日收益，不是累计 PnL）：

⚠️ 不要用累计 PnL 算相关。累计曲线自带强趋势，会把不同信号的相关性严重夸大。

7.3 回测

payload = {
    "type": "REGULAR",
    "settings": {
        "instrumentType": "EQUITY", "region": "USA", "universe": "TOP3000",
        "delay": 1, "decay": 0, "neutralization": "SUBINDUSTRY",
        "truncation": 0.08, "pasteurization": "ON", "unitHandling": "VERIFY",
        "nanHandling": "ON", "language": "FASTEXPR", "visualization": False,
    },
    "regular": "group_rank(ts_rank(operating_income/equity, 126), subindustry)",
}
resp = session.post("https://api.worldquantbrain.com/simulations", json=payload)
sim_id = resp.headers["Location"].rstrip("/").split("/")[-1]

while True:
    data = session.get(f"https://api.worldquantbrain.com/simulations/{sim_id}").json()
    if data.get("status") == "COMPLETE":
        alpha_id = data["alpha"]
        break
    time.sleep(8)

alpha = session.get(f"https://api.worldquantbrain.com/alphas/{alpha_id}").json()

7.4 提交与监控

# 提交
sub = session.post(f"https://api.worldquantbrain.com/alphas/{alpha_id}/submit")
print(sub.status_code)  # 201 成功

# 监控 SELF_CORRELATION
for _ in range(30):
    alpha = session.get(f"https://api.worldquantbrain.com/alphas/{alpha_id}").json()
    sc = next((c for c in alpha.get("is", {}).get("checks", []) if c["name"] == "SELF_CORRELATION"), {})
    if sc.get("result") in ("PASS", "FAIL"):
        break
    time.sleep(60)

7.5 自动提交模板

import numpy as np

def simulate_and_submit(expression, settings, existing_pnls=None):
    """
    existing_pnls: {alpha_id: [cum_pnl_values]}，已上线因子的累计 PnL 序列。
    返回: {"alpha_id": ..., "decision": "submitted|skip|high_corr|verify_failed", ...}
    """
    payload = {"type": "REGULAR", "settings": settings, "regular": expression}
    resp = session.post("https://api.worldquantbrain.com/simulations", json=payload)
    if resp.status_code != 201:
        return {"error": "simulate_failed"}
    sim_id = resp.headers["Location"].rstrip("/").split("/")[-1]
    while True:
        data = session.get(f"https://api.worldquantbrain.com/simulations/{sim_id}").json()
        if data.get("status") == "COMPLETE":
            alpha_id = data["alpha"]
            break
        if data.get("status") in ("ERROR", "FAILED"):
            return {"error": "simulation_error"}
        time.sleep(8)
    alpha = session.get(f"https://api.worldquantbrain.com/alphas/{alpha_id}").json()
    is_ = alpha.get("is", {})

    # 1. 基础指标过滤
    if is_.get("fitness", 0) < 1.1 or is_.get("sharpe", 0) < 1.3 or is_.get("turnover", 1) > 0.20:
        return {"alpha_id": alpha_id, "decision": "skip", "reason": "metrics", "metrics": is_}

    # 2. 相关性检查（基于日收益）
    def daily_rets(cum):
        return [cum[i+1] - cum[i] for i in range(len(cum) - 1)]

    if existing_pnls:
        new_pnl = fetch_pnl(session, alpha_id)
        new_ret = daily_rets(new_pnl)
        for old_id, old_pnl in existing_pnls.items():
            old_ret = daily_rets(old_pnl)
            if len(new_ret) == len(old_ret) and len(new_ret) > 20:
                corr = abs(float(np.corrcoef(new_ret, old_ret)[0, 1]))
                if corr >= 0.7:
                    # 例外：新 Sharpe 高于旧 Sharpe 10% 以上可提交
                    old_sharpe = None  # 需从外部传入或缓存
                    if old_sharpe is None or is_.get("sharpe", 0) < old_sharpe * 1.1:
                        return {"alpha_id": alpha_id, "decision": "high_corr", "corr_with": old_id, "corr": corr}

    # 3. 提交
    sub = session.post(f"https://api.worldquantbrain.com/alphas/{alpha_id}/submit")
    if sub.status_code not in (200, 201):
        return {"alpha_id": alpha_id, "decision": "submit_failed", "status": sub.status_code}

    # 4. 验证是否真正上线（BRAIN 可能因 SELF_CORRELATION 保持 UNSUBMITTED）
    for _ in range(20):
        time.sleep(10)
        alpha = session.get(f"https://api.worldquantbrain.com/alphas/{alpha_id}").json()
        if alpha.get("status") == "ACTIVE":
            return {"alpha_id": alpha_id, "decision": "submitted", "status": "ACTIVE"}
        sc = next((c for c in alpha.get("is", {}).get("checks", []) if c["name"] == "SELF_CORRELATION"), {})
        if sc.get("result") == "FAIL":
            return {"alpha_id": alpha_id, "decision": "self_correlation_fail", "status": alpha.get("status")}

    return {"alpha_id": alpha_id, "decision": "verify_failed", "status": alpha.get("status")}

7.6 限流

• 模拟/提交间 sleep 2–5 秒。
• 遇 429 读取 Retry-After，指数退避。
• 批量建议单线程或 ≤ 2 并发。

7.7 提交后验证（201 ≠ 已上线）

POST /alphas/{id}/submit 返回 201 只表示请求被接受，不代表 alpha 已变为 ACTIVE。实战中常见：

• alpha 状态仍为 UNSUBMITTED（SELF_CORRELATION 未通过或审核中）。
• 同一信号换参数生成的新 alpha被系统判定为重复，无法真正提交。

必须二次确认：

alpha = session.get(f"{API_BASE}/alphas/{alpha_id}").json()
print(alpha.get("status"))  # ACTIVE 才算真正提交成功

# 如果 status == UNSUBMITTED，查看 checks 中 SELF_CORRELATION 结果
for c in alpha.get("is", {}).get("checks", []):
    print(c["name"], c.get("result"), c.get("value"))

获取全部 alpha 并统计 ACTIVE 数量：

def get_all_alphas(session, limit=100):
    all_alphas = []
    offset = 0
    while True:
        data = session.get(f"{API_BASE}/users/self/alphas", params={"limit": limit, "offset": offset}).json()
        batch = data.get("results", data.get("alphas", []))
        if not batch:
            break
        all_alphas.extend(batch)
        if len(batch) < limit:
            break
        offset += limit
    return all_alphas

all_alphas = get_all_alphas(session)
active = [a for a in all_alphas if a.get("status") == "ACTIVE"]
print(f"total={len(all_alphas)}, ACTIVE={len(active)}")

8. 组合构建规则

8.1 diversified 组合示例

8.2 提交优先级

1. 高 Fitness（≥ 1.5）且低 TO（< 15%）
2. 来自不同信号簇
3. 若 SELF_CORRELATION 冲突，保留高 Fitness 版本

8.3 相关性的真相

对 ACTIVE alpha 的日收益做相关分析，发现：

• 同一信号簇内相关性极高：
  • 两个 open-close 反转 + OI/Equity 混合（权重不同）日收益相关 0.84
  • 两个分析师 EPS 相关 0.74
  • 两个杠杆/质量因子（-equity/assets vs liabilities/assets）相关 0.84
• 跨簇也不一定能分散：基于 scl12_buzz 的情绪 alpha 与基于 est_eps/close 的分析师 alpha 相关仍达 0.59–0.67。
• 累计 PnL 相关性严重失真： alpha 的累计 PnL 两两相关普遍 > 0.90，容易让人误以为所有因子都一样。

结论：

• 换窗口、换权重、换 neutralization 不能创造真正的低相关。
• 真正的低相关来自 完全不同的数据来源或经济逻辑（如：宏观事件、期权流、跨境、另类数据）。
• 在常规 USA TOP3000 基本面/价量/分析师池子里，"低相关" 往往是 0.3–0.6 的日收益相关，不要追求 0。

9. 提交前 Checklist

• 已获取 所有 alpha 列表（含 ACTIVE / UNSUBMITTED），不只是本次模拟
• 新因子与已有 ACTIVE alpha 日收益 相关性 < 0.7（或新 Sharpe ≥ 旧 Sharpe × 1.1）
• 相关性基于 日收益 计算，不是累计 PnL
• 字段已验证
• 模拟无报错
• Sharpe ≥ 1.3（理想 ≥ 1.5）
• Fitness ≥ 1.1
• Turnover 1%–20%（可放宽至 ≤ 35%）
• Drawdown < 15%
• 所有 IS 检查 PASS
• 多空数量合理
• 提交后 再次确认 status == ACTIVE，201 不代表上线

10. 核心经验（一句话版）

1. 先生成因子前先拉取所有 ACTIVE alpha 的 PnL，避免高相关重复。
2. 相关性必须算日收益，累计 PnL 相关会把所有因子看成同一个。
3. 201 响应 ≠ 提交成功：提交后必须确认 status == ACTIVE。
4. 基本面 > 混合 > 技术：operating_income/equity、est_eps/close、free_cash_flow_reported_value/equity 是最稳起点。
5. group_rank + ts_rank 是黄金组合。
6. SUBINDUSTRY 中性化通过率最高。
7. Decay 是控制换手的主杠杆：基本面 0，技术 10–30。
8. 50/50 正交混合能降低换手，但未必能降低相关；相关靠信号来源，不靠权重。
9. 字段先验证，无效字段秒级报错。
10. USA TOP3000 里真正的低相关很难做；同一数据池的 "不同" 表达式往往高度相关。

11. 自进化机制

每次与 BRAIN 交互（提交、查询、分析）后，AI 应把新发现写回本 SKILL，使其随实战经验持续进化。

11.1 触发条件

以下任一情况发生后，运行一次 scripts/evolve_skill.py：

• 提交了一个或多个新 alpha
• 批量回测了一批 alpha
• 查询了 alpha 状态并发现变化（如 UNSUBMITTED → ACTIVE，或被拒绝）
• 发现了新的字段可用性/失效模式

11.2 运行方式

前提：设置 WQ_BRAIN_USERNAME / WQ_BRAIN_PASSWORD，或在 skill 目录下放置未跟踪的 credential.txt，内容为 BRAIN 账号密码 JSON 数组：

["your_username", "your_password"]

# 1. 预览：生成建议追加的 markdown 片段，不修改任何文件
pyenv exec python scripts/evolve_skill.py

# 2. 提交：追加到 SKILL.md 并更新 alpha_db.json
pyenv exec python scripts/evolve_skill.py --apply

注意：不带 --apply 的预览模式不会修改 alpha_db.json 和 SKILL.md，你可以先审查再提交。脚本仅依赖 requests 和 numpy，不需要 wq-bus 项目代码。数据文件已随 SKILL 分发。

脚本会：

1. 拉取 /users/self/alphas（分页）获取全部 alpha。
2. 与本地 alpha_db.json 对比，找出 新增 或 状态/指标变化 的 alpha。
3. 对新 alpha 抓取 recordsets/pnl，计算与已有 ACTIVE alpha 的 日收益相关性。
4. 自动生成经验条目（指标评价 + 相关评价 + 表达式摘要）。
5. 第一次运行输出批量快照；后续运行输出增量条目。
6. --apply 模式下把条目追加到 ## 12. 实证记录（自动更新），并保存本地 alpha_db.json。

11.3 AI 应如何整理经验

脚本输出后，AI 需要人工判断哪些条目值得永久写入 SKILL：

• 保留：高 Fitness 低换手的成功案例、新的低相关信号簇、意外的失败模式。
• 精简：大量重复的同一信号簇条目应合并为一句话规律。
• 更新模板/阈值：如果多次发现某个字段/模板失效，应回到第 4、5、6 节更新。

11.4 数据结构

• alpha_db.json：本地 alpha 快照库，包含状态、指标、表达式、PnL。该文件会包含个人研究记录，默认被 .gitignore 忽略，不应提交到公开仓库。
• SKILL.md：最终人类可读 playbook，第 12 节只保留脱敏后的通用经验。

12. 实证记录（自动更新）

本节仅保留机制说明。真实运行生成的 alpha ID、表达式、PnL、提交状态和相关性记录可能关联个人账号与研究资产，默认写入本地 alpha_db.json，不随仓库发布。 若需要沉淀通用经验，请人工汇总成脱敏规则后再写回第 4、5、6、8、10 节。