diff --git a/README.md b/README.md index 58c6a7e..507e395 100644 --- a/README.md +++ b/README.md @@ -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)。 + + +## TODO + +增加1D级别的模型训练和预测 + ## 快速开始 运行环境需要 Python 3、Go、可访问的 banbot 源码和 banbot 配置中的 QuestDB。默认路径是： @@ -75,18 +81,20 @@ predictions/latest_v1/report.json 一键脚本生成的 CSV 固定只包含以下 10 列： -| 列 | 含义 | -| --- | --- | -| `rank` | 按 8H USDT 成交额从高到低的排名 | -| `symbol` | 合约符号 | -| `latest_bar_start_utc` | 最新完整 8H K 线的 UTC 起始时间 | -| `close` | 最新完整 8H K 线收盘价 | -| `quote_volume_usdt_8h` | 最新 8H 的 USDT 成交额 | -| `atr_rank_200_pct` | 当前 ATR 在过去 200 根 8H ATR 中的百分位，单位为 `%` | -| `drawdown_point_pct` | 点预测对应的回撤百分比 | -| `drawdown_cap_80pct_pct` | q80 对应的回撤百分比上界参考 | -| `price_level_reached_80pct` | q20 对应的价格水平参考 | -| `price_level_cap_80pct` | q80 对应的价格水平参考 | + +| 列 | 含义 | +| --------------------------- | ------------------------------------- | +| `rank` | 按 8H USDT 成交额从高到低的排名 | +| `symbol` | 合约符号 | +| `latest_bar_start_utc` | 最新完整 8H K 线的 UTC 起始时间 | +| `close` | 最新完整 8H K 线收盘价 | +| `quote_volume_usdt_8h` | 最新 8H 的 USDT 成交额 | +| `atr_rank_200_pct` | 当前 ATR 在过去 200 根 8H ATR 中的百分位，单位为 `%` | +| `drawdown_point_pct` | 点预测对应的回撤百分比 | +| `drawdown_cap_80pct_pct` | 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/model.md`](docs/model.md)：目标定义、特征、训练协议、指标和限制。 - [`docs/v0_0823.md`](docs/v0_0823.md)、[`docs/v1_0823.md`](docs/v1_0823.md)：历史实验记录，不是当前运行入口。 + diff --git a/cbpred/latest_predict.py b/cbpred/latest_predict.py index dd6994c..d713b6e 100644 --- a/cbpred/latest_predict.py +++ b/cbpred/latest_predict.py @@ -10,6 +10,7 @@ from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlparse +import joblib import pandas as pd 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]: 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]] = [] - total = len(manifest["symbols"]) - for index, item in enumerate(manifest["symbols"], 1): + total = len(prediction_items) + for index, item in enumerate(prediction_items, 1): csv_path = data_dir / item["path"] frame = pd.read_csv(csv_path).sort_values("timestamp").drop_duplicates("timestamp") latest = frame.iloc[-1] @@ -128,6 +148,7 @@ def build_predictions(data_dir: Path, artifact: Path, output_dir: Path) -> dict[ summary = { "model": str(artifact.resolve()), "symbols": len(rows), + "skipped_symbols": skipped_symbols, "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, "csv": str(output_csv.resolve()), diff --git a/tests/test_latest_predict.py b/tests/test_latest_predict.py index 47a5946..8e496ae 100644 --- a/tests/test_latest_predict.py +++ b/tests/test_latest_predict.py @@ -1,6 +1,12 @@ +import json +import tempfile 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): @@ -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__": unittest.main()