kyugao/sfgrid
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完成模型更新

Browse Source
This commit is contained in:
WIN-ALB39P9B6BU\Docker
2026-06-24 12:19:08 +08:00
parent ee8fc60eae
commit 6494f43ddd
22 changed files with 4029 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files
models/reference_backtest_logic/README.md
+158
View File
@@ -0,0 +1,158 @@
# v6.7r3 代码使用说明
`code/` 目录包含 v6.7r3 策略的**自包含**实现,可在不依赖项目其他模块的情况下独立运行。
## 文件清单
| 文件 | 大小 | 作用 |
|---|---|---|
| `strategy.py` | ~22 KB | 完整策略实现(模型加载/特征/网格/沉寂/周度淘汰) |
## 文件结构
```
strategy.py
├── 配置常量 (网格/价格/沉寂/周度)
├── 1. 持仓 + 网格交易
│ ├── Position dataclass
│ ├── compute_initial_position() 初始建仓
│ ├── compute_single_position() 单格建仓
│ └── simulate_grid_day() 单日网格 (LIFO)
├── 2. 5 特征沉寂检测
│ └── is_slumbering() 5 特征 ≥ 3 触发
├── 3. 模型加载 + 三件套预测
│ ├── ModelBundle class
│ │ ├── load 3 .pkl
│ │ └── predict(X56) → {rank_score, top_prob, stack_prob}
├── 4. 特征计算
│ ├── compute_52_base_features() 52 维 v3.4 基础
│ └── compute_4_v67_new_features() 4 维 v6.7 新增
├── 5. 评分池
│ └── score_pool() 单日全市场评分
├── 6. 主回测入口(精简版)
│ └── quick_backtest() 生产级完整版见 tools/backtest_v67r2.py
└── 7. 入口示例
└── __main__ 加载模型 + 加载行情 + 跑回测
```
## 快速开始
```bash
# 1. 安装依赖
pip install pandas numpy lightgbm scipy
# 2. 准备数据
# 方式 A: 用项目内的 dump_market_data_to_parquet.py 拉 Postgres
python tools/dump_market_data_to_parquet.py
# 方式 B: 直接用现有 parquet (release/v6.7r3 之前已生成 5025 个)
# 3. 运行
cd release/v6.7r3/code
python strategy.py
```
## 核心 API 速查
### 1. 加载模型
```python
from strategy import ModelBundle
models = ModelBundle("../models")
# 等价: models = ModelBundle("release/v6.7r3/models")
print(models.rank_feats[:5]) # ['rolling_grid_ratio_20d', ...]
```
### 2. 三件套预测
```python
import numpy as np
import pandas as pd
# 加载单只股 120 日窗口
df_window = pd.read_parquet("data/market_data/share/000001.parquet")
df_window["date"] = pd.to_datetime(df_window["date"])
df_window = df_window.tail(120).reset_index(drop=True)
# 算 56 维特征 (这里用简化版, 生产建议用项目 core.features)
from strategy import compute_52_base_features, compute_4_v67_new_features
fd = compute_52_base_features(df_window)
fd = compute_4_v67_new_features(df_window, fd)
X56 = np.array([fd.get(k, 0.0) for k in models.rank_feats], dtype=np.float64).reshape(1, -1)
# 三件套预测
pred = models.predict(X56)
print(f"rank_score: {pred['rank_score'][0]:.3f}")
print(f"top_prob: {pred['top_prob'][0]:.3f}")
print(f"stack_prob: {pred['stack_prob'][0]:.3f}") # 月末选股用
```
### 3. 沉寂检测
```python
from strategy import is_slumbering
df = pd.read_parquet("data/market_data/share/000001.parquet")
df["date"] = pd.to_datetime(df["date"])
slumbering = is_slumbering(df, min_triggers=3)
# True = 5 特征中至少 3 个触发 → 资金离场
```
### 4. 网格交易(单日)
```python
from strategy import Position, simulate_grid_day
pos = Position(code="000001", base=10, queue=[10.0, 9.5], entry_date="2023-05-01")
new_base, new_queue, trades = simulate_grid_day(
pos.base, pos.queue,
open_p=9.4, high_p=10.6, low_p=9.3, close_p=10.5,
)
# 触发 buy at 9.0 (low 9.3 <= 9.0? no, 9.3 > 9.0 不触发)
# 触发 sell at 10.0 (high 10.6 >= 10.0, 卖出 1 格)
# 最终: base=11, queue=[9.5] (1 格清仓)
```
### 5. 评分池(每日全市场)
```python
from strategy import score_pool, ModelBundle
import pandas as pd
# 假设 kline_cache 是 dict[code, DataFrame]
models = ModelBundle("../models")
pool = score_pool(pd.Timestamp("2024-03-15"), kline_cache, models, models.rank_feats)
# pool 列: code6, latest_close, rank_score, top_prob, stack_prob
# 按 stack_prob 降序, 取 top 10
top10 = pool.head(10)
```
## 关键参数(可调)
| 参数 | 默认 | 说明 | 调优方向 |
|---|---|---|---|
| `WEEKLY_ELIM_N` | 2 | 周度淘汰: 连续 N 周不在 top 50 | 1 太频, 3+ 太慢 |
| `SLUMBER_TRIGGERS` | 3 | 沉寂检测: >= 3/5 特征触发 | 2 太宽, 4 太严 |
| `SLUMBER_DAYS` | 10 | 连续触发多少天清仓 | 5 太短, 20 太长 |
| `TOP_N` | 10 | 最大持仓数 | 5-15 视资金量 |
| `SHARES_PER_GRID` | 200 | 单格股数 (2 手) | 100 (1 手) 也可 |
| `REFILL_PRICE` | 9.0-9.8 | 补仓价格区间 | 紧贴 9-10 网格上限 |
## 依赖项目其他模块?
为保持 `code/` 目录**自包含**:
- ✅ 不依赖 `core/features.py` (内置 `compute_52_base_features` 简化版)
- ✅ 不依赖 `training/dataset_builder.py` (内置 `compute_4_v67_new_features`)
- ❌ 需要数据: parquet 行情文件
**生产部署建议**:用 `core/features.calculate_features` 替换 `compute_52_base_features`,
它有完整版 v3.4 52 维特征实现,精度更高。
## 完整版 vs 精简版
| 维度 | `code/strategy.py` (精简) | `tools/backtest_v67r2.py` (生产) |
|---|---|---|
| 特征计算 | 简化版 (10 维示例) | 完整版 (52 维 v3.4 + 4 维 v6.7) |
| 数据加载 | dict of DataFrame | parquet 目录 + score cache |
| 日志 | 无 | 详细进度打印 |
| 输出 | dict 指标 | CSV/JSON/PNG 完整产物 |
| 速度 | 慢 (无缓存) | 快 (有 score cache) |
| 用途 | 教学/集成 | 完整回测 |
**生产环境**:用 `tools/backtest_v67r2.py` 跑回测,确保完整功能。
**集成到实盘系统**:用 `code/strategy.py` 中的 `ModelBundle` 和 `quick_backtest` 作为模板。
models/reference_backtest_logic/strategy.py
+574
View File
@@ -0,0 +1,574 @@
"""
v6.7r3 策略核心实现 —— 自包含版
================================
包含:
1. 模型加载
2. 56 维特征计算
3. 三件套预测 (rank → top → stacking)
4. 网格交易模拟器
5. 5 特征沉寂检测
6. 周度评分淘汰
7. 月末调仓 + 清仓补入
依赖:
pip install pandas numpy lightgbm scipy
(内嵌了核心算法, 不依赖项目其他模块, 可独立运行)
"""
from __future__ import annotations
import math
import pickle
import time
from collections import defaultdict
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import numpy as np
import pandas as pd
# ============================================================
# 0. 配置
# ============================================================
# 网格
INITIAL_CASH = 60_000.0
TOP_N = 10
TOP_MODEL_N = 50
SHARES_PER_GRID = 200
MIN_BUY_PRICE = 7.0
MAX_BUY_PRICE = 10.0
REFILL_MIN_PRICE = 9.0
REFILL_MAX_PRICE = 9.8
GRID_LOWER, GRID_UPPER = 1, 11
# 沉寂检测
SLUMBER_TRIGGERS = 3 # >= 3/5 特征触发
SLUMBER_DAYS = 10 # 连续 10 日触发
SLUMBER_LOOKBACK_60 = 60
SLUMBER_LOOKBACK_20 = 20
# 周度淘汰
WEEKLY_ELIM_N = 2 # 连续 2 周不在 top 50 → 卖
# ============================================================
# 1. 持仓 + 网格交易
# ============================================================
@dataclass
class Position:
code: str
base: int # 当前基准价 (整数)
queue: list = field(default_factory=list) # 持仓队列: 每格成本价
entry_date: str = ""
def compute_initial_position(close: float) -> Tuple[int, list]:
"""初始建仓: base=ceil(close), queue=[10, 9, ..., base] 且 >= close"""
base = math.ceil(close)
base = max(GRID_LOWER, min(base, GRID_UPPER))
queue = [g for g in range(GRID_UPPER, base - 1, -1) if g >= close]
return base, queue
def compute_single_position(close: float) -> Tuple[int, list]:
"""单格建仓"""
base = math.ceil(close)
base = max(GRID_LOWER, min(base, GRID_UPPER))
return base, [base]
def simulate_grid_day(base, queue, open_p, high_p, low_p, close_p, can_buy=True):
"""单日网格交易 (LIFO)"""
trades = []
new_base, new_queue = base, list(queue)
# 1) 买
buy_price = new_base - 1
if low_p <= buy_price and new_base > GRID_LOWER and can_buy:
new_base = buy_price
new_queue.append(buy_price)
trades.append({"direction": "buy", "price": buy_price, "shares": SHARES_PER_GRID, "pnl": 0.0})
# 2) 卖 (循环)
while True:
sell_price = new_base + 1
if high_p >= sell_price and new_queue:
buy_cost = new_queue.pop()
pnl = (sell_price - buy_cost) * SHARES_PER_GRID
trades.append({"direction": "sell", "price": sell_price, "shares": SHARES_PER_GRID, "pnl": pnl})
new_base = sell_price
else:
break
return new_base, new_queue, trades
# ============================================================
# 2. 5 特征沉寂检测
# ============================================================
def is_slumbering(df: pd.DataFrame, lookback_60=60, lookback_20=20,
min_triggers=SLUMBER_TRIGGERS) -> bool:
"""检测一只股是否陷入'沉寂' (资金离场后长期低位震荡).
5 特征, >= min_triggers 触发.
"""
if df is None or len(df) < lookback_60:
return False
sub = df.tail(lookback_60)
close = sub["close"].values
high = sub["high"].values
low = sub["low"].values
vol = sub["volume"].values
# 1. 波动率塌陷
log_ret = np.log(close[1:] / close[:-1])
if len(log_ret) < lookback_20:
return False
vol_20d = float(np.std(log_ret[-lookback_20:], ddof=1))
vol_60d = float(np.std(log_ret, ddof=1))
vol_collapse = (vol_60d > 0) and (vol_20d / vol_60d < 0.6)
# 2. 振幅萎缩
amp_20d = float(np.mean((high[-lookback_20:] - low[-lookback_20:]) / close[-lookback_20:]) * 100)
amp_shrink = amp_20d < 2.5
# 3. 成交量枯竭
avg_vol_20 = float(np.mean(vol[-lookback_20:]))
avg_vol_60 = float(np.mean(vol))
vol_dry = (avg_vol_60 > 0) and (avg_vol_20 / avg_vol_60 < 0.5)
# 4. 价格弱势
price_max_60 = float(np.max(close))
price_weak = price_max_60 > 0 and (close[-1] / price_max_60) < 0.85
# 5. 反弹失败
recent_high_30 = float(np.max(high[-30:]))
past_high_60 = float(np.max(high))
rebound_fail = past_high_60 > 0 and (recent_high_30 / past_high_60) < 0.95
triggers = [vol_collapse, amp_shrink, vol_dry, price_weak, rebound_fail]
return sum(triggers) >= min_triggers
# ============================================================
# 3. 模型加载 + 三件套预测
# ============================================================
class ModelBundle:
"""v6.7r3 三件套模型封装"""
def __init__(self, model_dir: str | Path):
model_dir = Path(model_dir)
with open(model_dir / "rank_lambdarank.pkl", "rb") as f:
rank_b = pickle.load(f)
with open(model_dir / "top_v67r2.pkl", "rb") as f:
top_b = pickle.load(f)
with open(model_dir / "stacking_v67r2.pkl", "rb") as f:
stack_b = pickle.load(f)
self.rank_model = rank_b["model"]
self.rank_feats = rank_b["feat_names"] # 56 维
self.top_model = top_b["model"]
self.top_feats = top_b["feat_names"] # 53 维
self.stack_model = stack_b["model"]
self.stack_feats = stack_b["feat_names"] # 55 维
@staticmethod
def _safe_predict_proba(model, X):
if hasattr(model, "predict_proba"):
return model.predict_proba(X)[:, 1]
return model.predict(X, raw_score=False)
def predict(self, X56: np.ndarray) -> dict:
"""输入 56 维特征矩阵 (n, 56), 返回三件套预测 dict.
返回: rank_score (n,), top_prob (n,), stack_prob (n,)
"""
rank_pred = self.rank_model.predict(X56)
# 52 维基础特征在 X56 中的索引
base_52_idx = [self.rank_feats.index(f) for f in self.top_feats if f != "rank_predicted_rounds"]
X53 = np.column_stack([X56[:, base_52_idx], rank_pred])
top_prob = self._safe_predict_proba(self.top_model, X53)
X55 = np.column_stack([X53, top_prob])
stack_prob = self._safe_predict_proba(self.stack_model, X55)
return {
"rank_score": rank_pred,
"top_prob": top_prob,
"stack_prob": stack_prob,
}
# ============================================================
# 4. 特征计算 (从项目 core/features.py 抽取, 关键函数)
# ============================================================
def _log_returns(close: np.ndarray) -> np.ndarray:
return np.log(close[1:] / close[:-1])
def _ema(arr: np.ndarray, period: int) -> np.ndarray:
"""指数移动平均"""
alpha = 2.0 / (period + 1)
out = np.zeros_like(arr)
out[0] = arr[0]
for i in range(1, len(arr)):
out[i] = alpha * arr[i] + (1 - alpha) * out[i-1]
return out
def compute_52_base_features(df: pd.DataFrame, total_share: Optional[float] = None) -> dict:
"""计算 v3.4 的 52 维基础特征 (简化版, 不完全等同于原版).
注: 完整版在 core/features.py, 这里用 pandas/numpy 简化.
"""
n = len(df)
if n < 60:
return None
close = df["close"].values
high = df["high"].values
low = df["low"].values
open_ = df["open"].values
vol = df["volume"].values
feats = {}
# 1. rolling_grid_ratio_20d
feats["rolling_grid_ratio_20d"] = float(np.mean((close[-20:] >= 1) & (close[-20:] <= 11)) * 100)
# 2. cross_freq_20d
diff = close[1:] - close[:-1]
sign_change = np.sum(np.abs(np.diff(np.sign(diff[-19:]))) > 0)
feats["cross_freq_20d"] = float(sign_change / 19 * 100) if 19 > 0 else 0.0
# 3. avg_daily_amp
feats["avg_daily_amp"] = float(np.mean((high - low) / open_) * 100) if n > 0 else 0.0
# 4. high_amp_days
feats["high_amp_days"] = float(np.mean((high - low) / open_ > 0.02) * 100) if n > 0 else 0.0
# 5. atr_pct
tr = np.maximum(high - low, np.maximum(np.abs(high - np.roll(close, 1)),
np.abs(low - np.roll(close, 1))))
feats["atr_pct"] = float(np.mean(tr[-14:]) / close[-1] * 100) if close[-1] > 0 else 0.0
# 6. volatility_20d (年化)
log_ret = _log_returns(close)
feats["volatility_20d"] = float(np.std(log_ret[-20:], ddof=1) * np.sqrt(252) * 100) if len(log_ret) >= 20 else 0.0
# 7. price_cv
feats["price_cv"] = float(np.std(close, ddof=1) / np.mean(close) * 100) if n > 1 and np.mean(close) > 0 else 0.0
# 8. bb_width (布林带宽)
ma20 = np.mean(close[-20:])
sd20 = np.std(close[-20:], ddof=1)
feats["bb_width"] = float((4 * sd20) / ma20 * 100) if ma20 > 0 else 0.0
# 9. volume_ratio
avg_vol_20 = float(np.mean(vol[-20:])) if n >= 20 else float(np.mean(vol))
avg_vol_60 = float(np.mean(vol[-60:])) if n >= 60 else float(np.mean(vol))
feats["volume_ratio"] = avg_vol_20 / avg_vol_60 if avg_vol_60 > 0 else 0.0
# 10. obv_slope
direction = np.sign(np.diff(close))
direction = np.concatenate([[0], direction])
obv = np.cumsum(direction * vol)
if len(obv) >= 20:
x = np.arange(20)
y = obv[-20:]
feats["obv_slope"] = float((np.polyfit(x, y, 1)[0]) / (np.mean(np.abs(y)) + 1e-10))
else:
feats["obv_slope"] = 0.0
# ... (其他 42 维特征省略, 完整版在 core/features.py)
# 这里只展示 10 个核心特征的计算模式
# 实际部署时建议直接调用项目 core.features.calculate_features
return feats
def compute_4_v67_new_features(df: pd.DataFrame, fd: dict) -> dict:
"""计算 v6.7 新增的 4 维特征"""
n = len(df)
if n < 60:
return fd
close = df["close"].values
high = df["high"].values
low = df["low"].values
# 53. vol_decay_5d
log_ret = np.log(close[1:] / close[:-1])
if len(log_ret) >= 20:
vol_5d = float(np.std(log_ret[-5:], ddof=1))
vol_20d = float(np.std(log_ret[-20:], ddof=1))
fd["vol_decay_5d"] = vol_5d / vol_20d if vol_20d > 0 else 0.0
else:
fd["vol_decay_5d"] = 0.0
# 54. grid_touch_relative_10d
if n >= 60:
range_10d = float(np.max(high[-10:]) - np.min(low[-10:]))
range_60d = float(np.max(high[-60:]) - np.min(low[-60:]))
close_now = float(close[-1])
close_60d_mean = float(np.mean(close[-60:]))
if close_now > 0 and close_60d_mean > 0 and range_60d > 0:
fd["grid_touch_relative_10d"] = (range_10d / close_now) / (range_60d / close_60d_mean)
else:
fd["grid_touch_relative_10d"] = 0.0
else:
fd["grid_touch_relative_10d"] = 0.0
# 55. vol_decay_x_grid_balance
fd["vol_decay_x_grid_balance"] = fd.get("vol_decay_5d", 0.0) * fd.get("grid_room_balance", 0.0)
# 56. vol_decay_x_dist_lower
fd["vol_decay_x_dist_lower"] = fd.get("vol_decay_5d", 0.0) * fd.get("dist_to_grid_lower", 0.0)
return fd
# ============================================================
# 5. 评分池(单日)
# ============================================================
def score_pool(date: pd.Timestamp,
kline_cache: Dict[str, pd.DataFrame],
models: ModelBundle,
feat_names: list) -> pd.DataFrame:
"""对所有有 120 日历史的股算 v6.7 三件套预测.
返回 DataFrame: code6, latest_close, rank_score, top_prob, stack_prob
"""
PRICE_MIN, PRICE_MAX = 1.0, 11.0
rows = []
codes = []
closes = []
for code, df in kline_cache.items():
sub = df[df["date"] <= date]
if len(sub) < 120:
continue
latest = float(sub["close"].iloc[-1])
if not (PRICE_MIN <= latest <= PRICE_MAX):
continue
# 算 56 维特征
obs = sub.tail(120).reset_index(drop=True)
fd = compute_52_base_features(obs)
if fd is None:
continue
fd = compute_4_v67_new_features(obs, fd)
try:
X = np.array([fd.get(k, 0.0) for k in feat_names], dtype=np.float64).reshape(1, -1)
pred = models.predict(X)
except Exception:
continue
rows.append(pred)
codes.append(code)
closes.append(latest)
if not rows:
return pd.DataFrame(columns=["code6", "latest_close", "rank_score", "top_prob", "stack_prob"])
import numpy as np
return pd.DataFrame({
"code6": codes,
"latest_close": closes,
"rank_score": [r["rank_score"][0] for r in rows],
"top_prob": [r["top_prob"][0] for r in rows],
"stack_prob": [r["stack_prob"][0] for r in rows],
}).sort_values("stack_prob", ascending=False).reset_index(drop=True)
# ============================================================
# 6. 主回测入口(精简版, 仅展示核心逻辑)
# ============================================================
def quick_backtest(kline_cache: Dict[str, pd.DataFrame],
models: ModelBundle,
start_date: str = "2023-05-01",
end_date: str = "2026-04-30",
weekly_elim_n: int = WEEKLY_ELIM_N,
slumber_days: int = SLUMBER_DAYS) -> dict:
"""精简版 3 年回测 (生产级完整版见 tools/backtest_v67r2.py).
核心流程:
1. 初始建仓
2. 每日: 网格交易 + 沉寂检测
3. 周五: 周度评分淘汰 + 补仓
4. 月末: 清仓补入 (触及 11 元)
Returns:
dict: 总收益率, 年化夏普, 最大回撤, 周胜率, 终值
"""
feat_names = models.rank_feats
# 交易日索引
sample = next(iter(kline_cache.values()))
all_dates = pd.DatetimeIndex(sorted(sample["date"].unique()))
mask = (all_dates >= start_date) & (all_dates <= end_date)
backtest_dates = all_dates[mask]
# 初始评分 (INIT_SELECT_DATE)
init_date = pd.Timestamp("2023-04-28")
init_pool = score_pool(init_date, kline_cache, models, feat_names)
init_pool = init_pool[init_pool["latest_close"].apply(lambda p: MIN_BUY_PRICE <= p <= MAX_BUY_PRICE)]
init_pool = init_pool.sort_values("stack_prob", ascending=False).head(TOP_N)
# 初始建仓
positions: Dict[str, Position] = {}
cash = INITIAL_CASH
for _, row in init_pool.iterrows():
code = row["code6"]
actual_close = float(row["latest_close"])
base, grid_queue = compute_initial_position(actual_close)
if not grid_queue:
continue
positions[code] = Position(code, base, [actual_close] * len(grid_queue), "2023-04-28")
for _ in grid_queue:
cash -= actual_close * SHARES_PER_GRID
# 周度淘汰历史
top50_history = defaultdict(list)
slumber_streak: Dict[str, int] = {}
total_asset_history = []
weekly_pnl = []
for i, cur_date in enumerate(backtest_dates):
cur_str = cur_date.strftime("%Y-%m-%d")
is_week_end = (i == len(backtest_dates) - 1) or (backtest_dates[i + 1].week != cur_date.week)
# === 网格日间交易 ===
for code, pos in list(positions.items()):
if not pos.queue:
continue
df = kline_cache[code]
sub = df[df["date"] == cur_date]
if sub.empty:
continue
row = sub.iloc[0]
new_base, new_queue, day_trades = simulate_grid_day(
pos.base, pos.queue,
float(row["open"]), float(row["high"]), float(row["low"]), float(row["close"]),
)
if day_trades:
pos.base, pos.queue = new_base, new_queue
for t in day_trades:
if t["direction"] == "buy":
cash -= t["price"] * t["shares"]
else:
cash += t["price"] * t["shares"]
# === 沉寂检测 ===
for code, pos in list(positions.items()):
if not pos.queue:
continue
df = kline_cache[code]
sub = df[df["date"] <= cur_date]
if len(sub) < 60:
continue
slumber = is_slumbering(sub)
slumber_streak[code] = slumber_streak.get(code, 0) + 1 if slumber else 0
if slumber_streak[code] >= slumber_days and pos.queue:
# 全仓清仓
px = float(sub["close"].iloc[-1])
cash += px * len(pos.queue) * SHARES_PER_GRID
positions[code] = Position(code, 0, [], cur_str)
slumber_streak[code] = 0
# === 资产快照 ===
mv = sum((float(kline_cache[c][kline_cache[c]["date"] <= cur_date]["close"].iloc[-1])
* len(p.queue) * SHARES_PER_GRID)
for c, p in positions.items() if p.queue)
total_asset = cash + mv
total_asset_history.append(total_asset)
# === 周度淘汰 + 补仓 ===
if is_week_end and i > 0 and weekly_elim_n > 0:
pool = score_pool(cur_date, kline_cache, models, feat_names)
top50 = set(pool.head(TOP_MODEL_N)["code6"].tolist())
for code in list(positions.keys()):
top50_history[code].append(code in top50)
# 连续 N 周不在 top 50 → 卖出
inactive = []
for code, p in positions.items():
if not p.queue:
continue
hist = top50_history.get(code, [])
if len(hist) >= weekly_elim_n and all(x is False for x in hist[-weekly_elim_n:]):
inactive.append(code)
for code in inactive[:1]: # 每月最多淘汰 1 只 (与 v6.3 一致)
pos = positions[code]
sub = kline_cache[code][kline_cache[code]["date"] <= cur_date]
px = float(sub["close"].iloc[-1])
cash += px * len(pos.queue) * SHARES_PER_GRID
positions[code] = Position(code, 0, [], cur_str)
# 补仓
positions = {c: p for c, p in positions.items() if p.queue}
refill_needed = max(0, TOP_N - len(positions))
if refill_needed > 0:
ref_pool = pool[pool["latest_close"].apply(lambda p: REFILL_MIN_PRICE < p < REFILL_MAX_PRICE)]
ref_pool = ref_pool[~ref_pool["code6"].isin(positions.keys())]
ref_pool = ref_pool.head(refill_needed)
for _, row in ref_pool.iterrows():
code = row["code6"]
actual_close = float(row["latest_close"])
base, grid_queue = compute_single_position(actual_close)
if not grid_queue:
continue
cost = actual_close * SHARES_PER_GRID * len(grid_queue)
if cash < cost:
continue
positions[code] = Position(code, base, [actual_close] * len(grid_queue), cur_str)
cash -= cost
# 计算指标
final_value = total_asset_history[-1] if total_asset_history else INITIAL_CASH
total_return = final_value / INITIAL_CASH - 1
rets = np.diff(total_asset_history) / total_asset_history[:-1]
sharpe = float(rets.mean() / rets.std() * np.sqrt(52)) if len(rets) > 1 and rets.std() > 0 else 0.0
cum_max = np.maximum.accumulate(total_asset_history)
dd = (np.array(total_asset_history) - cum_max) / cum_max
max_dd = float(dd.min())
win_rate = float((rets > 0).mean()) if len(rets) > 0 else 0.0
return {
"total_return_pct": round(total_return * 100, 2),
"annual_sharpe": round(sharpe, 4),
"max_drawdown_pct": round(max_dd * 100, 2),
"weekly_win_rate_pct": round(win_rate * 100, 2),
"final_value": round(final_value, 2),
}
# ============================================================
# 7. 入口示例
# ============================================================
if __name__ == "__main__":
# 1. 加载模型
models = ModelBundle("models") # 默认从当前目录的 models/ 加载
print(f"✓ 加载模型: rank {len(models.rank_feats)} 维, "
f"top {len(models.top_feats)} 维, stack {len(models.stack_feats)} 维")
# 2. 加载行情 (示例: 从 parquet 目录)
# 实际部署时, 从 market_data.kline_stock (Postgres) 或本地 parquet 加载
from pathlib import Path
parquet_dir = Path("data/market_data/share")
kline_cache = {}
for p in parquet_dir.glob("*.parquet"):
df = pd.read_parquet(p)
df["date"] = pd.to_datetime(df["date"])
kline_cache[p.stem] = df
print(f"✓ 加载行情: {len(kline_cache)} 只股")
# 3. 跑精简版回测
metrics = quick_backtest(kline_cache, models)
print("\n=== v6.7r3 三年回测结果 (精简版) ===")
for k, v in metrics.items():
print(f" {k}: {v}")