对冲驱动满仓化（Hedge-Driven Full Position）Implementation Plan

> For Claude: Use ${SUPERPOWERS_SKILLS_ROOT}/skills/collaboration/executing-plans/SKILL.md to implement this plan task-by-task.

Goal: 当某锚定指数的择时建议低仓位/有做空信号时，被释放的资金不再按评分顺位递补股指ETF，而是通过"负相关性+长期上涨"的数据驱动模型，自动分配给全池中最适合对冲该风险的品种（国债、红利、黄金、货币等），目标95%满仓。同时将锚定置信度改为数值展示、增加做空信号展示区。

Architecture: 核心改造点是在 selection/ 下新增 hedge_allocator.py 模块，替代现有 _compute_portfolio_weights 中的简单等权+区域截断逻辑。算法三步：(1)为每个锚点计算 risk_on_ratio 和 hedge_demand（含做空加成），(2)预计算"锚点→全池ETF"7×N相关矩阵，(3)先做风险配置再按 hedge_score 分配对冲资金。同时修复diversification.py的收益率切片bug、在engine.py输出中增加corr_to_anchor数值列、在show_marker.py增加做空信号展示区。

Tech Stack: Python 3.12, pandas, numpy, 现有5因子打分体系, SQLite (cross_section_portfolio表新增列)

---

Task 1: 新增锚点→全池相关矩阵计算模块

Files:

• Create: adata/aohan/quant/selection/factor/anchor_correlation.py
• Test: adata/aohan/quant/selection/test_anchor_correlation.py

这个模块计算7个锚定指数与全池所有ETF的60日收益率相关系数矩阵，供hedge_allocator使用。

Step 1: 写测试

# adata/aohan/quant/selection/test_anchor_correlation.py
import numpy as np
import pandas as pd
from adata.aohan.quant.selection.factor.anchor_correlation import AnchorCorrelationMatrix

def test_correlation_matrix_shape():
    dates = pd.date_range('2025-01-01', periods=120, freq='B')
    anchor_prices = {
        '399006.SZ': pd.Series(100 + np.cumsum(np.random.randn(120)*0.5), index=dates),
        'QQQ': pd.Series(200 + np.cumsum(np.random.randn(120)*0.8), index=dates),
    }
    etf_prices = {
        '511090.SH': pd.Series(100 + np.cumsum(np.random.randn(120)*0.1), index=dates),
        '510300.SH': anchor_prices['399006.SZ'] * 1.01 + np.random.randn(120)*0.05,
    }
    matrix = AnchorCorrelationMatrix(window=60)
    corr = matrix.compute(anchor_prices, etf_prices)
    assert corr.shape == (2, 2)
    assert -1 <= corr.loc['399006.SZ', '510300.SH'] <= 1
    assert corr.loc['399006.SZ', '510300.SH'] > 0.9  # 高度相关
    print("correlation matrix test passed")

if __name__ == '__main__':
    test_correlation_matrix_shape()

Step 2: 运行确认失败（模块不存在）

Run: env -u PYTHONHOME -u PYTHONPATH .venv/bin/python adata/aohan/quant/selection/test_anchor_correlation.py Expected: ModuleNotFoundError

Step 3: 实现 AnchorCorrelationMatrix

# adata/aohan/quant/selection/factor/anchor_correlation.py
import numpy as np
import pandas as pd


class AnchorCorrelationMatrix:
    def __init__(self, window: int = 60):
        self.window = window
        self._last_matrix = None

    def compute(self, anchor_prices: dict, etf_prices: dict) -> pd.DataFrame:
        """
        计算锚定指数与全池ETF的相关系数矩阵。
        
        Parameters
        ----------
        anchor_prices : dict[str, pd.Series]
            {anchor_code: 价格序列（日期索引）}
        etf_prices : dict[str, pd.Series]
            {etf_code: 价格序列（日期索引）}
            
        Returns
        -------
        pd.DataFrame : index=anchor_codes, columns=etf_codes, values=Pearson corr
        """
        anchor_rets = {}
        for code, prices in anchor_prices.items():
            s = prices.iloc[-(self.window + 1):]
            rets = s.pct_change().dropna()
            if len(rets) >= self.window // 2:
                anchor_rets[code] = rets

        etf_rets = {}
        for code, prices in etf_prices.items():
            s = prices.iloc[-(self.window + 1):]
            rets = s.pct_change().dropna()
            if len(rets) >= self.window // 2:
                etf_rets[code] = rets

        result = pd.DataFrame(index=list(anchor_rets.keys()), columns=list(etf_rets.keys()), dtype=float)
        for a_code, a_rets in anchor_rets.items():
            for e_code, e_rets in etf_rets.items():
                common_idx = a_rets.index.intersection(e_rets.index)
                if len(common_idx) < 20:
                    result.loc[a_code, e_code] = np.nan
                    continue
                result.loc[a_code, e_code] = np.corrcoef(
                    a_rets.loc[common_idx].values,
                    e_rets.loc[common_idx].values
                )[0, 1]

        self._last_matrix = result
        return result

    def get_last(self) -> pd.DataFrame:
        return self._last_matrix

注意收益率计算用 pct_change().dropna() 而不是 np.diff/切片，避免重蹈 diversification.py 的覆辙。

Step 4: 运行测试通过

Run: env -u PYTHONHOME -u PYTHONPATH .venv/bin/python adata/aohan/quant/selection/test_anchor_correlation.py Expected: "correlation matrix test passed"

Step 5: Commit

git add adata/aohan/quant/selection/factor/anchor_correlation.py adata/aohan/quant/selection/test_anchor_correlation.py
git commit -m "feat: 新增AnchorCorrelationMatrix计算锚点→全池ETF相关矩阵"

---

Task 2: 新增 HedgeAllocator 核心分配模块

Files:

• Create: adata/aohan/quant/selection/allocator/__init__.py
• Create: adata/aohan/quant/selection/allocator/hedge_allocator.py
• Test: adata/aohan/quant/selection/test_hedge_allocator.py

Step 1: 写测试（覆盖核心场景）

# adata/aohan/quant/selection/test_hedge_allocator.py
import pandas as pd
import numpy as np
from adata.aohan.quant.selection.allocator.hedge_allocator import HedgeAllocator, HedgeConfig


def _make_test_data():
    """构造测试数据：3个锚点、5只ETF、模拟价格"""
    dates = pd.date_range('2025-01-01', periods=120, freq='B')
    np.random.seed(42)
    # 创业板：熊市，low timing
    cyb = 100 + np.cumsum(-np.abs(np.random.randn(120))*0.3 - 0.1)
    # 纳指：牛市，high timing
    nq = 200 + np.cumsum(np.random.randn(120)*0.5 + 0.05)
    # 国债：慢牛
    bond = 100 + np.cumsum(np.random.randn(120)*0.05 + 0.01)
    
    anchor_prices = {'399006.SZ': pd.Series(cyb, index=dates), 'QQQ': pd.Series(nq, index=dates)}
    etf_prices = {
        '159915.SZ': pd.Series(cyb*1.01, index=dates),  # 创业板ETF - 跟399006高相关
        '513100.SH': pd.Series(nq*1.01, index=dates),    # 纳指ETF - 跟QQQ高相关
        '511090.SH': pd.Series(bond, index=dates),        # 30年国债 - 跟399006低相关
        '518880.SH': pd.Series(bond*0.9 + 50, index=dates),  # 黄金 - 中等相关
    }
    etf_info = {
        '159915.SZ': {'anchor': '399006.SZ', 'region': 'A股本土', 'category': '宽基指数', 'factor_score': 0.8},
        '513100.SH': {'anchor': 'QQQ', 'region': '美股', 'category': '跨境', 'factor_score': 0.75},
        '511090.SH': {'anchor': None, 'region': '防御', 'category': '债券', 'factor_score': 0.2},
        '518880.SH': {'anchor': 'GLD', 'region': '防御', 'category': '商品', 'factor_score': 0.5},
    }
    timing = {
        '399006.SZ': {'today': 10, 'yesterday': 20, 'predicted': 15},  # 低仓位
        'QQQ': {'today': 90, 'yesterday': 85, 'predicted': 90},        # 高仓位
        'short-399006.SZ': {'today': 20},                               # 做空20%
        'short-GLD': {'today': 0},
    }
    return anchor_prices, etf_prices, etf_info, timing


def test_bear_market_hedge_allocation():
    """场景：创业板timing=10% + short=20%，应将资金对冲到国债"""
    anchor_prices, etf_prices, etf_info, timing = _make_test_data()
    config = HedgeConfig()
    allocator = HedgeAllocator(config)
    result = allocator.allocate(
        ranked_etfs=list(etf_info.keys()),
        etf_info=etf_info,
        anchor_prices=anchor_prices,
        etf_prices=etf_prices,
        timing_positions=timing,
        premium_coefs={c: 1.0 for c in etf_info},
    )
    weights = result['weights']
    # 创业板ETF应该只拿到约10%仓位
    assert weights['159915.SZ'] < 0.2, f"创业板ETF应低权重, got {weights['159915.SZ']}"
    # 纳指ETF应该接近满权重
    assert weights['513100.SH'] > 0.3, f"纳指ETF应高权重, got {weights['513100.SH']}"
    # 国债应该被分配到对冲仓位（不为0）
    assert weights['511090.SH'] > 0.05, f"国债应有对冲仓位, got {weights['511090.SH']}"
    # 总仓位不超过95%
    total = sum(weights.values())
    assert total <= 0.97, f"总仓位不超95%+误差, got {total}"
    print(f"bear market test passed: weights={weights}, total={total:.2%}")


def test_bull_market_no_hedge():
    """场景：所有锚点timing高，应该基本不配防御品"""
    anchor_prices, etf_prices, etf_info, timing = _make_test_data()
    timing['399006.SZ'] = {'today': 95, 'yesterday': 90, 'predicted': 95}
    timing['short-399006.SZ'] = {'today': 0}
    config = HedgeConfig()
    allocator = HedgeAllocator(config)
    result = allocator.allocate(
        ranked_etfs=list(etf_info.keys()),
        etf_info=etf_info,
        anchor_prices=anchor_prices,
        etf_prices=etf_prices,
        timing_positions=timing,
        premium_coefs={c: 1.0 for c in etf_info},
    )
    weights = result['weights']
    # 国债在牛市不应被大量配入
    assert weights['511090.SH'] < 0.05, f"牛市国债应接近0, got {weights['511090.SH']}"
    total = sum(weights.values())
    print(f"bull market test passed: weights={weights}, total={total:.2%}")


if __name__ == '__main__':
    test_bear_market_hedge_allocation()
    test_bull_market_no_hedge()

Step 2: 运行确认失败

Run: env -u PYTHONHOME -u PYTHONPATH .venv/bin/python adata/aohan/quant/selection/test_hedge_allocator.py Expected: ModuleNotFoundError

Step 3: 实现 HedgeAllocator

# adata/aohan/quant/selection/allocator/__init__.py
from .hedge_allocator import HedgeAllocator, HedgeConfig

# adata/aohan/quant/selection/allocator/hedge_allocator.py
"""
对冲驱动的仓位分配器。

核心算法：
1. 为每个锚点计算 risk_on_ratio（留在本锚点ETF的资金比例）和 hedge_demand（需要对冲出去的资金）
2. risk_on 部分：按5因子评分排序分配给对应锚点的ETF，应用trend/premium系数
3. hedge 部分：计算每个候选ETF的 hedge_score = corr_negative × momentum × own_timing × premium_coef
   按score从高到低分配对冲资金，受单只上限/区域上限/总仓位约束
4. 兜底：货币基金 → 现金
"""
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple

from adata.aohan.quant.selection.factor.anchor_correlation import AnchorCorrelationMatrix
from adata.aohan.quant.utils.constants import CODE_PREFIX_SHORT


@dataclass
class HedgeConfig:
    target_total_weight: float = 0.95
    single_etf_cap: float = 0.15
    defensive_etf_cap: float = 0.20
    bear_timing_threshold: float = 0.30       # timing < 30% 视为空仓期
    hedge_corr_threshold: float = 0.30        # corr > 0.30 不能当对冲品
    min_score_threshold: float = 0.35
    correlation_window: int = 60
    long_term_momentum_days: int = 250
    min_long_term_return: float = 0.0         # 对冲品长期收益必须>0
    defensive_categories: tuple = ('债券', '商品', 'REITs')
    cash_etfs: tuple = ('511990.SH', '511880.SH')  # 货币基金兜底
    short_demand_amplifier: float = 1.0       # 做空加成系数（short_w × amplifier）
    region_caps: dict = field(default_factory=lambda: {
        'A股本土': 0.40, '港股': 0.30, '美股': 0.40,
        '日股': 0.20, '欧亚太': 0.20, '防御': 0.40,
    })


class HedgeAllocator:
    def __init__(self, config: HedgeConfig = None):
        self.config = config or HedgeConfig()
        self.corr_matrix_calc = AnchorCorrelationMatrix(window=self.config.correlation_window)

    def allocate(
        self,
        ranked_etfs: List[str],
        etf_info: Dict[str, dict],
        anchor_prices: Dict[str, pd.Series],
        etf_prices: Dict[str, pd.Series],
        timing_positions: Dict[str, dict],
        premium_coefs: Dict[str, float],
        trend_ratios: Optional[Dict[str, float]] = None,
    ) -> dict:
        cfg = self.config
        if trend_ratios is None:
            trend_ratios = {c: 1.0 for c in ranked_etfs}

        # Step 1: 计算每个锚点的 risk_on 和 hedge_demand
        anchor_demands = self._compute_anchor_demands(timing_positions)

        # Step 2: 计算锚点→全池相关矩阵
        corr_df = self.corr_matrix_calc.compute(anchor_prices, etf_prices)

        # Step 3: 计算每只ETF的长期动量
        lt_momentum = self._compute_long_term_momentum(etf_prices)

        # Step 4: 风险配置（risk-on allocation）
        weights = {c: 0.0 for c in ranked_etfs}
        region_used = {r: 0.0 for r in cfg.region_caps}
        total_used = 0.0
        hedge_pools = {a: 0.0 for a in anchor_demands}

        # 每个锚点的基础目标权重（等权分配给该锚点的ETF池，但受hedge_demand调节）
        anchor_base_weights = self._calc_anchor_base_weights(etf_info, ranked_etfs)

        for etf_code in ranked_etfs:
            info = etf_info[etf_code]
            anchor = info.get('anchor')
            region = info.get('region', 'A股本土')
            factor_score = info.get('factor_score', 0.5)

            if factor_score < cfg.min_score_threshold:
                break

            prem_c = premium_coefs.get(etf_code, 1.0)
            if prem_c < 0.05:
                continue
            trend_c = trend_ratios.get(etf_code, 1.0)

            if anchor is None:
                continue  # 无锚点的（如货币）在risk-on阶段跳过，在hedge阶段处理

            demand = anchor_demands.get(anchor)
            if demand is None:
                risk_on = 1.0
                hedge_d = 0.0
            else:
                risk_on = demand['risk_on_ratio']
                hedge_d = demand['hedge_demand']

            base_w = anchor_base_weights.get(anchor, 0.0) / max(1, sum(
                1 for c in ranked_etfs if etf_info[c].get('anchor') == anchor
                and etf_info[c].get('factor_score', 0) >= cfg.min_score_threshold
            ))

            adjusted_w = base_w * risk_on * trend_c * prem_c
            single_cap = cfg.defensive_etf_cap if info.get('category') in cfg.defensive_categories else cfg.single_etf_cap
            available = min(
                cfg.region_caps.get(region, 0.20) - region_used[region],
                cfg.target_total_weight - total_used,
                single_cap
            )
            actual_w = max(0, min(adjusted_w, available))
            if actual_w > 0.01:
                weights[etf_code] = actual_w
                region_used[region] += actual_w
                total_used += actual_w
            hedge_pools[anchor] += base_w * hedge_d

        # Step 5: 对冲配置（hedge allocation）
        for anchor, pool_size in hedge_pools.items():
            if pool_size <= 0.01:
                continue
            self._allocate_hedge_pool(
                anchor=anchor, pool_size=pool_size, corr_df=corr_df,
                weights=weights, region_used=region_used, total_used_ref={'v': total_used},
                etf_info=etf_info, etf_prices=etf_prices, lt_momentum=lt_momentum,
                timing_positions=timing_positions, premium_coefs=premium_coefs,
                trend_ratios=trend_ratios,
            )
            total_used = sum(weights.values())

        # Step 6: 兜底 - 货币基金
        remaining = cfg.target_total_weight - sum(weights.values())
        if remaining > 0.02:
            for cash_code in cfg.cash_etfs:
                if cash_code not in etf_info:
                    continue
                cash_w = min(remaining, 0.10)
                weights[cash_code] = weights.get(cash_code, 0.0) + cash_w
                remaining -= cash_w
                if remaining <= 0.01:
                    break

        return {
            'weights': weights,
            'anchor_demands': anchor_demands,
            'corr_matrix': corr_df,
            'total_weight': sum(weights.values()),
            'cash_remaining': 1.0 - sum(weights.values()),
            'hedge_pools': hedge_pools,
        }

    def _compute_anchor_demands(self, timing_positions) -> dict:
        """为每个锚点计算 risk_on_ratio 和 hedge_demand"""
        demands = {}
        processed = set()
        for key, tp in timing_positions.items():
            if key.startswith(CODE_PREFIX_SHORT):
                continue
            today = tp.get('today', 0)
            if today is None:
                today = 0
            long_w = today / 100.0
            risk_on = max(0.0, min(1.0, long_w))

            short_key = CODE_PREFIX_SHORT + key
            short_tp = timing_positions.get(short_key, {})
            short_today = short_tp.get('today', 0) or 0
            short_w = short_today / 100.0
            hedge_d = (1.0 - risk_on) + short_w * self.config.short_demand_amplifier
            hedge_d = max(0.0, min(1.0, hedge_d))

            demands[key] = {
                'long_w': long_w,
                'short_w': short_w,
                'risk_on_ratio': risk_on,
                'hedge_demand': hedge_d,
            }
            processed.add(key)
        return demands

    def _calc_anchor_base_weights(self, etf_info, ranked_etfs) -> dict:
        """计算每个锚点的基础目标权重（按有多少个锚点等权）"""
        anchors = set()
        for c in ranked_etfs:
            a = etf_info[c].get('anchor')
            if a:
                anchors.add(a)
        if not anchors:
            return {}
        w = 1.0 / len(anchors)
        return {a: w for a in anchors}

    def _compute_long_term_momentum(self, etf_prices) -> dict:
        """计算每只ETF的长期动量（近N日收益率）"""
        momentum = {}
        for code, prices in etf_prices.items():
            if len(prices) < 20:
                momentum[code] = 0.0
                continue
            lookback = min(self.config.long_term_momentum_days, len(prices) - 1)
            ret = prices.iloc[-1] / prices.iloc[-(lookback + 1)] - 1
            momentum[code] = ret
        return momentum

    def _allocate_hedge_pool(
        self, anchor, pool_size, corr_df, weights, region_used, total_used_ref,
        etf_info, etf_prices, lt_momentum, timing_positions, premium_coefs, trend_ratios
    ):
        cfg = self.config
        candidates = []
        for etf_code, info in etf_info.items():
            if etf_code not in corr_df.columns:
                continue
            corr_val = corr_df.loc[anchor, etf_code] if anchor in corr_df.index else np.nan
            if np.isnan(corr_val):
                continue
            if corr_val > cfg.hedge_corr_threshold:
                continue  # 相关性太高不能做对冲

            # 长期上涨检查
            lt_ret = lt_momentum.get(etf_code, 0)
            if lt_ret < cfg.min_long_term_return:
                continue

            # 自己择时检查
            own_anchor = info.get('anchor')
            own_timing = 1.0
            if own_anchor is not None:
                ot = timing_positions.get(own_anchor, {}).get('today')
                if ot is None:
                    own_timing = 0.5  # 无数据时给中性
                else:
                    own_timing = ot / 100.0
                if own_timing < cfg.bear_timing_threshold and own_anchor != anchor:
                    continue  # 自己也在空仓期不能当别人对冲

            prem_c = premium_coefs.get(etf_code, 1.0)
            if prem_c < 0.05:
                continue

            corr_neg = -corr_val  # 越负分数越高
            # 归一化到[0, 2]区间（corr从-1到0.3 → corr_neg从1到-0.3 → 截断到0以上）
            corr_score = max(0, corr_neg + 1) / 2  # 映射到[0,1]
            # 长期动量归一化
            mom_score = max(0, min(1, lt_ret / 0.3))  # 30%年涨幅算满分

            hedge_score = corr_score * (0.5 + 0.5 * mom_score) * own_timing * prem_c
            candidates.append((etf_code, hedge_score, info))

        candidates.sort(key=lambda x: -x[1])

        remaining = pool_size
        for etf_code, score, info in candidates:
            if remaining <= 0.01:
                break
            region = info.get('region', '防御')
            category = info.get('category', '')
            single_cap = cfg.defensive_etf_cap if category in cfg.defensive_categories else cfg.single_etf_cap
            current_w = weights.get(etf_code, 0)
            available = min(
                remaining,
                cfg.region_caps.get(region, 0.20) - region_used.get(region, 0),
                cfg.target_total_weight - sum(weights.values()),
                single_cap - current_w,
            )
            if available > 0.01:
                alloc = available * min(1.0, score)
                alloc = min(alloc, available)
                weights[etf_code] = current_w + alloc
                region_used[region] = region_used.get(region, 0) + alloc
                remaining -= alloc

Step 4: 运行测试通过

Run: env -u PYTHONHOME -u PYTHONPATH .venv/bin/python adata/aohan/quant/selection/test_hedge_allocator.py Expected: 两个测试都 passed

Step 5: Commit

git add adata/aohan/quant/selection/allocator/__init__.py adata/aohan/quant/selection/allocator/hedge_allocator.py adata/aohan/quant/selection/test_hedge_allocator.py
git commit -m "feat: 新增HedgeAllocator对冲驱动仓位分配器"

---

Task 3: ETFInfo 增加字段 + 抽取公共锚点映射

Files:

• Modify: adata/aohan/quant/selection/universe/etf_pool.py
• Create: adata/aohan/quant/selection/anchor_map.py（抽取公共锚点映射，消除engine和show_marker的重复）
• Modify: adata/aohan/quant/selection/engine.py（改import）
• Modify: adata/aohan/quant/chart_test/show_marker.py（改import）

Step 1: 修改 ETFInfo 增加字段

在 etf_pool.py 的 ETFInfo dataclass 中增加：

is_defensive: bool = False  # 是否防御品种（国债/货币/黄金等长期低相关品种）

在 CURATED_ETF_POOL 列表中，为债券类、黄金ETF(518880)、货币基金(511990/511880) 设置 is_defensive=True。

Step 2: 创建公共锚点映射 anchor_map.py

# adata/aohan/quant/selection/anchor_map.py
"""
公共锚定指数映射（消除 engine.py 和 show_marker.py 的重复代码）
"""
from typing import Optional, Tuple

ANCHOR_NAMES = {
    '399300.SZ': '沪深300',
    '399006.SZ': '创业板指',
    '399102.SZ': '创业板综',
    '399001.SZ': '深证成指',
    '000852.SH': '中证1000',
    'HSI': '恒生指数',
    'QQQ': '纳斯达克',
    'SPY': '标普500',
    'GLD': '黄金',
    '510880': '红利',
    '513050': '中概互联',
    '508068.SH': 'REITs',
}

# ETF代理：当锚点指数本身不在价格数据中时，用ETF代理获取价格
ANCHOR_PROXY = {
    '399102.SZ': '159908.SZ',
    '399006.SZ': '159915.SZ',
    '399300.SZ': '510300.SH',
    'QQQ': '513100.SH',
    'SPY': '513500.SH',
    'HSI': '159920.SZ',
    'GLD': '518880.SH',
    '510880': '510880.SH',
    '513050': '513050.SH',
    '508068.SH': '508056.SH',
}


def get_anchor(etf_info) -> str:
    """根据ETF的category/sub_category/underlying_index返回锚定指数代码。
    
    替代 engine._get_anchor_and_timing 中 _get_anchor 部分和 show_marker._get_anchor 的重复逻辑。
    """
    category = etf_info.category
    sub_category = etf_info.sub_category or ''
    underlying = etf_info.underlying_index or ''
    code = etf_info.ts_code

    if '399102' in underlying:
        return '399102.SZ'
    if category == '宽基指数':
        if sub_category in ('大盘', '超大盘', '中盘'):
            return '399300.SZ'
        if sub_category == '成长':
            return '399006.SZ'
        return '399102.SZ'
    if category == '行业主题':
        if sub_category == '科技':
            return '399006.SZ'
        return '399102.SZ'
    if category == '策略风格':
        return '510880'
    if category == '跨境':
        if sub_category in ('港股', '港股红利'):
            return 'HSI'
        if sub_category == '中概':
            return '513050'
        return 'QQQ'
    if category == '债券':
        return None  # 债券无锚点，由对冲系统数据驱动
    if category == '商品':
        return 'GLD'
    if category == 'REITs':
        return '508068.SH'
    return '399102.SZ'


def resolve_timing(anchor: str, timing_positions: dict) -> Tuple[Optional[float], Optional[float], Optional[float]]:
    """从timing_positions中解析锚点的today/yesterday/predicted，带fallback。"""
    if anchor is None:
        return None, None, None
    for key in [anchor, anchor.replace('.SZ','').replace('.SH','')]:
        if key in timing_positions:
            tp = timing_positions[key]
            return tp.get('today'), tp.get('yesterday'), tp.get('predicted')
    # fallback遍历前缀
    for key, tp in timing_positions.items():
        if key.startswith(anchor) or anchor.startswith(key):
            return tp.get('today'), tp.get('yesterday'), tp.get('predicted')
    return None, None, None

注意：债券类返回 None，表示债券没有锚点择时，完全由对冲数据驱动。这是关键改动。

Step 3: 修改 engine.py 的 _get_anchor_and_timing

让它调用新的公共函数：

from adata.aohan.quant.selection.anchor_map import get_anchor, resolve_timing

def _get_anchor_and_timing(self, code, timing_positions):
    etf_info = self.etf_pool.get_by_code(code)
    if etf_info is None:
        return '399102.SZ', None, None, None
    anchor = get_anchor(etf_info)
    if anchor is None:
        return None, None, None, None
    today, yesterday, predicted = resolve_timing(anchor, timing_positions)
    return anchor, today, yesterday, predicted

Step 4: 修改 show_marker.py 的 _get_anchor

from adata.aohan.quant.selection.anchor_map import get_anchor as _public_get_anchor

# 将原来的 _get_anchor 内部函数改为调用公共函数

Step 5: 验证不报错

Run: env -u PYTHONHOME -u PYTHONPATH .venv/bin/python -c "from adata.aohan.quant.selection.anchor_map import get_anchor, resolve_timing, ANCHOR_NAMES, ANCHOR_PROXY; print('anchor_map OK')" Expected: "anchor_map OK"

Step 6: Commit

git add adata/aohan/quant/selection/universe/etf_pool.py adata/aohan/quant/selection/anchor_map.py adata/aohan/quant/selection/engine.py adata/aohan/quant/chart_test/show_marker.py
git commit -m "refactor: 抽取公共锚点映射到anchor_map.py，债券无锚点由对冲驱动；ETFInfo增加is_defensive字段"

---

Task 4: 改造 engine.py 的 _compute_portfolio_weights 使用 HedgeAllocator

Files:

• Modify: adata/aohan/quant/selection/engine.py
• Test: 运行现有 weight_test.py 验证不崩溃

Step 1: 在 engine.__init__ 中初始化 HedgeAllocator

from adata.aohan.quant.selection.allocator import HedgeAllocator, HedgeConfig

# 在 __init__ 中：
self.hedge_allocator = HedgeAllocator(HedgeConfig())

Step 2: 重构 _compute_portfolio_weights

核心改动：

• 废弃全局 exposure_cap 乘法
• 收集 anchor_prices 和 etf_prices（从 build_snapshot 的数据中）
• 构造 etf_info dict 传给 allocator
• 调用 allocator.allocate() 获取 weights
• 把结果写入 DataFrame，新增列：anchor_confidence（数值相关系数）、is_hedge_position（是否对冲仓位）
• 保留原有列名兼容

具体实现要点：

• 在 run_selection 时 snapshot 数据已经有每只ETF的close数据，需要同时收集锚定指数（用ANCHOR_PROXY代理）的close数据
• 从 timing_positions 传入 allocator
• allocator 返回的 weights 直接作为最终权重，替代原来"建议权重×cap"流程
• 区域上限由 allocator 内部处理，不需要外层再 _rebalance_by_region_caps（但可保留作为最终安全检查）
• 从 corr_matrix 中提取每只ETF与其anchor的相关系数，写入 anchor_confidence 列

Step 3: 废弃 SimpleRegimeDetector 的 exposure_cap 全局控制

在 run_selection 中，regime 检测保留（用于推送文案），但 exposure_cap 不再参与权重计算。portfolio DataFrame 中的 exposure_cap 列可固定为1.0或保留 regime 值仅供参考。

Step 4: 运行现有验证脚本

Run: env -u PYTHONHOME -u PYTHONPATH .venv/bin/python adata/aohan/quant/selection/weight_test.py Expected: 能正常运行，输出权重分配结果，不崩溃。检查输出中债券/黄金类是否在空仓期有分配。

Step 5: Commit

git add adata/aohan/quant/selection/engine.py
git commit -m "feat: engine.py使用HedgeAllocator，废弃全局exposure_cap，新增anchor_confidence数值列"

---

Task 5: DB 表新增列

Files:

• Modify: adata/aohan/quant/db/cross_section.py

Step 1: 在 cross_section_portfolio 表建表SQL中新增列

在 L88-129 的 CREATE TABLE 中增加：

anchor_confidence REAL,        -- ETF与锚定指数的60日相关系数
is_hedge_position INTEGER DEFAULT 0,  -- 是否对冲仓位（1=是）
hedge_from_anchor TEXT,        -- 是从哪个锚点的对冲池分来的

同时在 init_cross_section_tables 中加列迁移（ALTER TABLE ADD COLUMN），兼容已有DB。

Step 2: 在 save_weekly_snapshot 中写入新列

INSERT_PF 的 columns 和 values 都加上这3个新列。

Step 3: Commit

git add adata/aohan/quant/db/cross_section.py
git commit -m "feat: portfolio表新增anchor_confidence/is_hedge_position/hedge_from_anchor列"

---

Task 6: show_marker.py 对齐 - 三层仓位段适配 + 做空信号展示

Files:

• Modify: adata/aohan/quant/chart_test/show_marker.py

Step 1: 适配新的权重体系

原来的 actual_pos = adj_weight × timing_coef × trend_ratio × prem_coef 三层乘法现在由 HedgeAllocator 内部处理了。show_marker.py 读取portfolio表时，weight 列已经是最终建议仓位，不需要再乘系数。

Step 2: 锚定置信度展示改为数值+颜色

在推送消息的持仓表中，将原来没有的置信度列加上，显示格式：

• corr ≥ 0.85: 绿色 0.92
• 0.60 ≤ corr < 0.85: 黄色 0.73
• corr < 0.60: 红色 0.41
• corr < 0: 紫色 -0.15

HTML 版本用 inline CSS 背景色。文本版本用 emoji + 数值。

Step 3: 新增"🎯 做空信号"展示区

在持仓表下方，读取 timing_positions['short-399102.SZ'] 和 timing_positions['short-GLD']（可能还有 short-QQQ），如果 today > 5%，显示：

🎯 做空信号（策略建议）
• 创业板做空：XX% — 对冲创业板指下行风险
• 黄金做空：XX% — 对冲黄金回调风险

只有 today > 5% 才显示该品种；都不满足则整个区块不显示。

Step 4: 对冲仓位标注

对于 is_hedge_position=1 的品种，在名称后加 🛡️ 盾牌图标，hover或注释说明"对冲仓位（由空仓期负相关性驱动）"。

Step 5: 验证推送消息生成

Run section engine's generate_push_message to verify formatting doesn't crash.

Step 6: Commit

git add adata/aohan/quant/chart_test/show_marker.py
git commit -m "feat: show_marker适配HedgeAllocator输出，置信度改为数值展示，新增做空信号区和对冲仓位🛡️标注"

---

Task 7: 历史回测验证（关键！）

Files:

• Modify: adata/aohan/quant/selection/backtest/ 下的回测器
• Create: adata/aohan/quant/selection/test_hedge_backtest.py（回测验证脚本）

Step 1: 用历史数据验证关键场景

写脚本回测以下时段，对比旧方案（固定Top10+exposure_cap）和新方案（对冲递补）：

• 2022年4月A股暴跌
• 2022年10月全球熊市
• 2024年1~2月A股持续阴跌
• 2025年4月关税战波动

验证指标：

• 熊市时国债/黄金是否真的被自动配入
• 总仓位是否接近95%
• 最大回撤是否小于旧方案
• 牛市时防御品权重是否回归低位

Step 2: 检查相关性实际数值

脚本输出每个锚点与国债ETF(511090)、黄金ETF(518880)、红利ETF(510880)的实际相关系数，验证"红利与科技负相关"、"国债与创业板负相关"等假设是否成立。

Step 3: Commit（测试脚本）

git add adata/aohan/quant/selection/test_hedge_backtest.py
git commit -m "test: 对冲递补回测验证脚本，覆盖2022熊市/2024阴跌/2025波动期"

---

Task 8: 发布到公网验证

Step 1: 重启服务

cd /Users/aohan/PycharmProjects/adata
env -u PYTHONHOME -u PYTHONPATH bash restart.sh

Step 2: 检查 https://quant.aohan.net 截面页面

验证：

• 持仓表中是否有"锚定置信度"数值列
• 是否有"🎯 做空信号"区块（当前数据若有做空信号）
• 熊市品种（如创业板相关）的权重是否被压缩
• 国债/黄金是否在对冲位置
• 总仓位显示是否接近95%

Step 3: 运行一次截面选基验证

env -u PYTHONHOME -u PYTHONPATH .venv/bin/python -m adata.aohan.quant.selection.run_selection

检查输出日志和DB中数据是否正确。

---

实施顺序总结

关键注意事项

• diversification.py L287 的收益率计算确认无bug：经实际验证 np.diff(close[-(w+1):]) / close[-(w+1):-1] 是正确的（:-1 是排除最后一个元素，不是倒序步长），结果与 pct_change 一致，无需修改
• 债券类ETF不再锚定399300.SZ做反向——旧逻辑是 timing_coef = max(0, 1-timing_coef) 的硬编码反向，新方案债券返回 anchor=None，由数据（相关性）自动驱动，不需要这个特殊处理
• 防御品（债券/黄金/REITs）不过最低评分阈值——它们5因子分天然低（动量/夏普差），在hedge阶段不应用min_score_threshold，只有risk-on阶段才用
• 货币基金(511990/511880)特殊处理——不过任何过滤，作为最终兜底
• 相关性计算用 pct_change()——AnchorCorrelationMatrix 统一使用 pct_change().dropna() 计算收益率，安全无切片陷阱
• show_marker.py的ANCHOR_PROXY需要扩展——国债/货币不需要ANCHOR_PROXY（它们自己就是ETF可直接取价）；但注意黄金(GLD)和REITs(508068)在anchor_map中保留代理映射
• 保持向后兼容——portfolio表中原来的weight列仍然是最终权重，前端不需要大改
• 红利ETF(510880)保持锚定510880自身——它有自己的择时策略，但在hedge阶段如果它与某个空仓锚点（如创业板）的相关性<0.3且自身长期正收益，也会被选为对冲品——这符合你说的"红利和科技可能形成负相关"