aszerW
6a5ae8efbf
Previously hardcoded equal weight (1/select_num), ignoring config weight type. Now reads position_weights from last daily_record, correctly showing rank-based weights.
129 lines
4.6 KiB
Python
129 lines
4.6 KiB
Python
"""分析 2022年4月底~5月初 CL=F 入选原因"""
|
|
import os, sys, math
|
|
from pathlib import Path
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from rotation.simple_rotation import SimpleRotationStrategy, slope_r2_score
|
|
|
|
if 'FLASK_API_URL' not in os.environ:
|
|
os.environ['FLASK_API_URL'] = 'https://k3s.tokenpluse.xyz'
|
|
|
|
strategy = SimpleRotationStrategy()
|
|
strategy._preload_data()
|
|
|
|
# 分析日期范围
|
|
start = pd.Timestamp('2022-04-15')
|
|
end = pd.Timestamp('2022-05-10')
|
|
n_days = strategy.config.factor.n_days # 25
|
|
|
|
print("=" * 80)
|
|
print(f" 分析 CL=F 动量信号 ({start.date()} ~ {end.date()})")
|
|
print(f" 窗口长度: {n_days} 天")
|
|
print("=" * 80)
|
|
|
|
# 获取所有 signal_codes 的 score
|
|
signal_codes = strategy.signal_codes
|
|
date_range = pd.bdate_range(start, end)
|
|
|
|
for date in date_range:
|
|
scores = {}
|
|
for code in signal_codes:
|
|
if code not in strategy.index_data:
|
|
continue
|
|
df = strategy.index_data[code]
|
|
mask = df.index <= date
|
|
recent = df.loc[mask]
|
|
if len(recent) < n_days:
|
|
continue
|
|
prices = recent['close'].values[-n_days:]
|
|
score = slope_r2_score(prices)
|
|
scores[code] = (score, prices[-1])
|
|
|
|
if not scores:
|
|
continue
|
|
|
|
# 排序
|
|
ranked = sorted(scores.items(), key=lambda x: x[1][0], reverse=True)
|
|
cl_rank = None
|
|
for i, (code, (score, price)) in enumerate(ranked):
|
|
if code == 'CL=F':
|
|
cl_rank = i + 1
|
|
break
|
|
|
|
cl_score = scores.get('CL=F', (None, None))[0]
|
|
cl_price = scores.get('CL=F', (None, None))[1]
|
|
|
|
print(f"\n{date.strftime('%Y-%m-%d')} | CL=F score={cl_score:.4f}, price={cl_price:.2f}, rank={cl_rank}/{len(ranked)}")
|
|
print(f" Top 5:")
|
|
for i, (code, (score, price)) in enumerate(ranked[:5]):
|
|
marker = " <<<" if code == 'CL=F' else ""
|
|
print(f" #{i+1} {code:<15} score={score:>10.4f} price={price:.2f}{marker}")
|
|
|
|
# 详细分析 CL=F 价格走势
|
|
print(f"\n{'='*80}")
|
|
print(f" CL=F 价格走势 (2022年3月~5月)")
|
|
print(f"{'='*80}")
|
|
|
|
df_cl = strategy.index_data['CL=F']
|
|
mask = (df_cl.index >= '2022-03-01') & (df_cl.index <= '2022-05-15')
|
|
cl_prices = df_cl.loc[mask, 'close']
|
|
|
|
for date, price in cl_prices.items():
|
|
# 计算25天窗口的score
|
|
mask2 = df_cl.index <= date
|
|
recent = df_cl.loc[mask2]
|
|
if len(recent) < n_days:
|
|
continue
|
|
prices = recent['close'].values[-n_days:]
|
|
score = slope_r2_score(prices)
|
|
normalized = prices / prices[0]
|
|
slope, intercept = np.polyfit(np.arange(len(normalized)), normalized, 1)
|
|
y_pred = slope * np.arange(len(normalized)) + intercept
|
|
ss_res = np.sum((normalized - y_pred) ** 2)
|
|
ss_tot = np.sum((normalized - np.mean(normalized)) ** 2)
|
|
r2 = 1 - ss_res / ss_tot if ss_tot > 0 else 0
|
|
|
|
flag = ""
|
|
if date.strftime('%Y-%m-%d') in ('2022-04-29', '2022-05-05'):
|
|
flag = " <<< 入选日"
|
|
|
|
print(f" {date.strftime('%Y-%m-%d')} price={price:>8.2f} score={score:>10.4f} "
|
|
f"slope={slope:>8.5f} R²={r2:.4f}{flag}")
|
|
|
|
# 分析 CL=F 的组内竞争
|
|
print(f"\n{'='*80}")
|
|
print(f" CL=F 所在组: 查看组内竞争")
|
|
print(f"{'='*80}")
|
|
|
|
groups = strategy.config.asset_pools.by_group
|
|
for group_name, assets in groups.items():
|
|
group_codes = [a.signal_source for a in assets.values()]
|
|
if 'CL=F' in group_codes:
|
|
print(f" 组名: {group_name}")
|
|
print(f" 组成员: {group_codes}")
|
|
|
|
# 4/29 和 5/5 的组内得分
|
|
for target_date_str in ['2022-04-29', '2022-05-05']:
|
|
target_date = pd.Timestamp(target_date_str)
|
|
print(f"\n {target_date_str} 组内得分:")
|
|
for code in group_codes:
|
|
if code not in strategy.index_data:
|
|
continue
|
|
df = strategy.index_data[code]
|
|
mask = df.index <= target_date
|
|
recent = df.loc[mask]
|
|
if len(recent) < n_days:
|
|
print(f" {code:<15} 数据不足")
|
|
continue
|
|
prices = recent['close'].values[-n_days:]
|
|
score = slope_r2_score(prices)
|
|
marker = " <<< TOP1" if score == max(
|
|
slope_r2_score(strategy.index_data[c].loc[strategy.index_data[c].index <= target_date]['close'].values[-n_days:])
|
|
for c in group_codes if c in strategy.index_data and len(strategy.index_data[c].loc[strategy.index_data[c].index <= target_date]) >= n_days
|
|
) else ""
|
|
print(f" {code:<15} score={score:>10.4f}{marker}")
|