fix predict

This commit is contained in:
2026-08-24 19:10:32 +08:00
parent 286626dfe5
commit 1f72dfdc1e
3 changed files with 96 additions and 15 deletions

View File

@@ -4,6 +4,12 @@
当前正式模型只有一个:[`artifacts/v1/model.joblib`](artifacts/v1/model.joblib)。它由最近 18 根 8H K 线预测未来 6 根 8H K 线的稳健低点收益,并输出点预测以及用于区间参考的 q20/q80 结果。根目录不放置历史实验模型;历史实验说明保留在 [`docs/v0_0823.md`](docs/v0_0823.md) 和 [`docs/v1_0823.md`](docs/v1_0823.md)。 当前正式模型只有一个:[`artifacts/v1/model.joblib`](artifacts/v1/model.joblib)。它由最近 18 根 8H K 线预测未来 6 根 8H K 线的稳健低点收益,并输出点预测以及用于区间参考的 q20/q80 结果。根目录不放置历史实验模型;历史实验说明保留在 [`docs/v0_0823.md`](docs/v0_0823.md) 和 [`docs/v1_0823.md`](docs/v1_0823.md)。
## TODO
增加1D级别的模型训练和预测
## 快速开始 ## 快速开始
运行环境需要 Python 3、Go、可访问的 banbot 源码和 banbot 配置中的 QuestDB。默认路径是: 运行环境需要 Python 3、Go、可访问的 banbot 源码和 banbot 配置中的 QuestDB。默认路径是:
@@ -75,18 +81,20 @@ predictions/latest_v1/report.json
一键脚本生成的 CSV 固定只包含以下 10 列: 一键脚本生成的 CSV 固定只包含以下 10 列:
| 列 | 含义 |
| --- | --- | | 列 | 含义 |
| `rank` | 按 8H USDT 成交额从高到低的排名 | | --------------------------- | ------------------------------------- |
| `symbol` | 合约符号 | | `rank` | 按 8H USDT 成交额从高到低的排名 |
| `latest_bar_start_utc` | 最新完整 8H K 线的 UTC 起始时间 | | `symbol` | 合约符号 |
| `close` | 最新完整 8H K 线收盘价 | | `latest_bar_start_utc` | 最新完整 8H K 线的 UTC 起始时间 |
| `quote_volume_usdt_8h` | 最新 8H 的 USDT 成交额 | | `close` | 最新完整 8H K 线收盘价 |
| `atr_rank_200_pct` | 当前 ATR 在过去 200 根 8H ATR 中的百分位,单位为 `%` | | `quote_volume_usdt_8h` | 最新 8H 的 USDT 成交额 |
| `drawdown_point_pct` | 点预测对应的回撤百分比 | | `atr_rank_200_pct` | 当前 ATR 在过去 200 根 8H ATR 中的百分位,单位为 `%` |
| `drawdown_cap_80pct_pct` | q80 对应的回撤百分比上界参考 | | `drawdown_point_pct` | 点预测对应的回撤百分比 |
| `price_level_reached_80pct` | q20 对应的价格水平参考 | | `drawdown_cap_80pct_pct` | q80 对应的回撤百分比上界参考 |
| `price_level_cap_80pct` | q80 对应的价格水平参考 | | `price_level_reached_80pct` | q20 对应的价格水平参考 |
| `price_level_cap_80pct` | q80 对应的价格水平参考 |
回撤列是收益的相反数:预测收益为负时表示价格可能低于当前收盘;预测收益为正时,回撤列可能为负,表示未来稳健低点仍高于当前收盘。模型不满足资格条件时,预测字段为空,不应把空值当作零回撤。 回撤列是收益的相反数:预测收益为负时表示价格可能低于当前收盘;预测收益为正时,回撤列可能为负,表示未来稳健低点仍高于当前收盘。模型不满足资格条件时,预测字段为空,不应把空值当作零回撤。
@@ -107,3 +115,4 @@ predictions/latest_v1/report.json
- [`docs/usage.md`](docs/usage.md):安装、数据流、命令、目录和故障排查。 - [`docs/usage.md`](docs/usage.md):安装、数据流、命令、目录和故障排查。
- [`docs/model.md`](docs/model.md):目标定义、特征、训练协议、指标和限制。 - [`docs/model.md`](docs/model.md):目标定义、特征、训练协议、指标和限制。
- [`docs/v0_0823.md`](docs/v0_0823.md)、[`docs/v1_0823.md`](docs/v1_0823.md):历史实验记录,不是当前运行入口。 - [`docs/v0_0823.md`](docs/v0_0823.md)、[`docs/v1_0823.md`](docs/v1_0823.md):历史实验记录,不是当前运行入口。

View File

@@ -10,6 +10,7 @@ from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from urllib.parse import urlparse from urllib.parse import urlparse
import joblib
import pandas as pd import pandas as pd
from .pipeline import load_database_url, parse_time_ms, predict_from_csv, prepare_data from .pipeline import load_database_url, parse_time_ms, predict_from_csv, prepare_data
@@ -84,9 +85,28 @@ def ensure_database(
def build_predictions(data_dir: Path, artifact: Path, output_dir: Path) -> dict[str, object]: def build_predictions(data_dir: Path, artifact: Path, output_dir: Path) -> dict[str, object]:
manifest = json.loads((data_dir / "manifest.json").read_text()) manifest = json.loads((data_dir / "manifest.json").read_text())
artifact_data = joblib.load(artifact)
trained_symbols = set(artifact_data.get("asset_profiles", {}))
if trained_symbols:
skipped_symbols = [
item["symbol"]
for item in manifest["symbols"]
if item["symbol"] not in trained_symbols
]
prediction_items = [
item for item in manifest["symbols"] if item["symbol"] in trained_symbols
]
else:
skipped_symbols = []
prediction_items = manifest["symbols"]
if skipped_symbols:
print(
f"[latest-predict] skipped {len(skipped_symbols)} symbols without stored training profiles",
flush=True,
)
rows: list[dict[str, object]] = [] rows: list[dict[str, object]] = []
total = len(manifest["symbols"]) total = len(prediction_items)
for index, item in enumerate(manifest["symbols"], 1): for index, item in enumerate(prediction_items, 1):
csv_path = data_dir / item["path"] csv_path = data_dir / item["path"]
frame = pd.read_csv(csv_path).sort_values("timestamp").drop_duplicates("timestamp") frame = pd.read_csv(csv_path).sort_values("timestamp").drop_duplicates("timestamp")
latest = frame.iloc[-1] latest = frame.iloc[-1]
@@ -128,6 +148,7 @@ def build_predictions(data_dir: Path, artifact: Path, output_dir: Path) -> dict[
summary = { summary = {
"model": str(artifact.resolve()), "model": str(artifact.resolve()),
"symbols": len(rows), "symbols": len(rows),
"skipped_symbols": skipped_symbols,
"eligible_predictions": sum(row["drawdown_point_pct"] is not None for row in rows), "eligible_predictions": sum(row["drawdown_point_pct"] is not None for row in rows),
"latest_bar_start_utc": rows[0]["latest_bar_start_utc"] if rows else None, "latest_bar_start_utc": rows[0]["latest_bar_start_utc"] if rows else None,
"csv": str(output_csv.resolve()), "csv": str(output_csv.resolve()),

View File

@@ -1,6 +1,12 @@
import json
import tempfile
import unittest import unittest
from pathlib import Path
from unittest.mock import patch
from cbpred.latest_predict import OUTPUT_COLUMNS import pandas as pd
from cbpred.latest_predict import OUTPUT_COLUMNS, build_predictions
class LatestPredictTests(unittest.TestCase): class LatestPredictTests(unittest.TestCase):
@@ -21,6 +27,51 @@ class LatestPredictTests(unittest.TestCase):
), ),
) )
def test_batch_skips_symbols_without_training_profiles(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
data_dir = root / "data"
output_dir = root / "output"
data_dir.mkdir()
for symbol, filename in (
("TRAIN/USDT:USDT", "train.csv"),
("STALE/USDT:USDT", "stale.csv"),
):
pd.DataFrame(
[{"timestamp": 1, "quote": 10.0, "close": 100.0}]
).to_csv(data_dir / filename, index=False)
(data_dir / "manifest.json").write_text(
json.dumps(
{
"symbols": [
{"symbol": "TRAIN/USDT:USDT", "path": "train.csv"},
{"symbol": "STALE/USDT:USDT", "path": "stale.csv"},
]
}
)
)
result = {
"symbol": "TRAIN/USDT:USDT",
"timestamp": 1,
"close": 100.0,
"volatility_rank_200": 0.5,
"eligible_for_model": False,
}
with patch(
"cbpred.latest_predict.joblib.load",
return_value={"asset_profiles": {"TRAIN/USDT:USDT": {}}},
), patch(
"cbpred.latest_predict.predict_from_csv", return_value=result
) as predict:
summary = build_predictions(data_dir, root / "model.joblib", output_dir)
predict.assert_called_once_with(
root / "model.joblib", data_dir / "train.csv", symbol="TRAIN/USDT:USDT"
)
self.assertEqual(summary["symbols"], 1)
self.assertEqual(summary["skipped_symbols"], ["STALE/USDT:USDT"])
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()