From c0df80154b5a637116c2f8f39ea9a19ccc21b0b4 Mon Sep 17 00:00:00 2001 From: anyongjin Date: Tue, 25 Aug 2026 09:23:30 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=811d=E7=BA=A7=E5=88=AB=E6=8F=92?= =?UTF-8?q?=E9=92=88=E6=A8=A1=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 34 ++++++++---- cbpred/__main__.py | 13 +++-- cbpred/experiments.py | 36 ++++++++++--- cbpred/latest_predict.py | 22 ++++++-- cbpred/pipeline.py | 100 +++++++++++++++++++++++++++-------- cmd/cbpred-export/main.go | 19 +++---- docs/1d_2026-08-24.md | 77 +++++++++++++++++++++++++++ docs/model.md | 8 +++ docs/usage.md | 19 +++++-- tests/test_artifact_paths.py | 2 + tests/test_latest_predict.py | 5 +- tests/test_pipeline.py | 29 ++++++++++ 12 files changed, 299 insertions(+), 65 deletions(-) create mode 100644 docs/1d_2026-08-24.md diff --git a/README.md b/README.md index 60b20ad..b0308fb 100644 --- a/README.md +++ b/README.md @@ -1,15 +1,11 @@ # cbpred -`cbpred` 是一个面向 Binance U 本位 USDT 永续合约的统一 8H K 线回撤预测管线。数据由 [banbot](../banbot) 下载，模型只使用 OHLCV，不使用资金费率、盘口、持仓量或外部行情。 +`cbpred` 是一个面向 Binance U 本位 USDT 永续合约的统一 K 线回撤预测管线，默认正式模型使用 8H 周期。数据由 [banbot](../banbot) 下载，模型只使用 OHLCV，不使用资金费率、盘口、持仓量或外部行情。 -当前正式模型只有一个：[`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 结果。1d 模型作为独立实验产物保存在 [`artifacts_1d/best/model.joblib`](artifacts_1d/best/model.joblib)，不会覆盖默认 8h 模型；实验报告见 [`docs/1d_2026-08-24.md`](docs/1d_2026-08-24.md)。 -## TODO - -增加1D级别的模型训练和预测 - ## 快速开始 运行环境需要 Python 3、Go、可访问的 banbot 源码和 banbot 配置中的 QuestDB。默认路径是： @@ -81,6 +77,24 @@ predictions/latest_v1/report.json 如果数据已经由 banbot 更新完成，可在 `prepare` 中加入 `--skip-download`，只执行筛选、导出和整理。 +### 1d 实验 + +1d 使用相同的 `compact6` 特征、ExtraTrees 点模型、q20/q80 分位数模型、过滤和训练切分协议，只有 K 线周期改为 1d： + +```bash +.venv/bin/python -m cbpred prepare \ + --timeframe 1d \ + --out-dir data_1d \ + --banbot-config /home/ctyun/ban/banbot/biz/config.yml +.venv/bin/python -m cbpred.experiments \ + --data-dir data_1d \ + --output-dir artifacts_1d \ + --variants compact6 \ + --n-estimators 20 +``` + +结果目录为 `data_1d/`、`artifacts_1d/` 和 `predictions/latest_1d/`。1d 批量预测的成交额列名为 `quote_volume_usdt_1d`。 + ## 预测 CSV 一键脚本生成的 CSV 固定只包含以下 10 列： @@ -88,11 +102,11 @@ predictions/latest_v1/report.json | 列 | 含义 | | --------------------------- | ------------------------------------- | -| `rank` | 按 8H USDT 成交额从高到低的排名 | +| `rank` | 按请求周期 USDT 成交额从高到低的排名 | | `symbol` | 合约符号 | -| `latest_bar_start_utc` | 最新完整 8H K 线的 UTC 起始时间 | -| `close` | 最新完整 8H K 线收盘价 | -| `quote_volume_usdt_8h` | 最新 8H 的 USDT 成交额 | +| `latest_bar_start_utc` | 最新完整请求周期 K 线的 UTC 起始时间 | +| `close` | 最新完整请求周期 K 线收盘价 | +| `quote_volume_usdt_8h` / `quote_volume_usdt_1d` | 最新请求周期的 USDT 成交额 | | `atr_rank_200_pct` | 当前 ATR 在过去 200 根 8H ATR 中的百分位，单位为 `%` | | `drawdown_point_pct` | 点预测对应的回撤百分比 | | `drawdown_cap_80pct_pct` | q80 对应的回撤百分比上界参考 | diff --git a/cbpred/__main__.py b/cbpred/__main__.py index ad547db..41e9a26 100644 --- a/cbpred/__main__.py +++ b/cbpred/__main__.py @@ -14,16 +14,17 @@ from .pipeline import ( def build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description="Train a unified 8H futures drawdown model") + parser = argparse.ArgumentParser(description="Train a unified futures drawdown model") sub = parser.add_subparsers(dest="command", required=True) - prepare = sub.add_parser("prepare", help="filter symbols, call banbot, export complete 8H CSVs") + prepare = sub.add_parser("prepare", help="filter symbols, call banbot, export complete timeframe CSVs") prepare.add_argument("--out-dir", type=Path, default=Path("data")) prepare.add_argument("--banbot-dir", type=Path, default=Path("/home/ctyun/ban/banbot")) prepare.add_argument("--banbot-config", type=Path, default=Path("/home/ctyun/ban/banbot/biz/config.yml")) prepare.add_argument("--dsn", default=None) prepare.add_argument("--db-type", choices=["auto", "questdb", "timescale"], default="auto") prepare.add_argument("--exchange", default="binance") + prepare.add_argument("--timeframe", choices=["8h", "1d"], default="8h") prepare.add_argument("--end", dest="end_ms", default=None, help="UTC date/ISO timestamp or milliseconds") prepare.add_argument("--history-months", type=int, default=36) prepare.add_argument("--holdout-months", type=int, default=10) @@ -55,7 +56,7 @@ def build_parser() -> argparse.ArgumentParser: run.add_argument("--volatility-rank-threshold", type=float, default=VOLATILITY_RANK_THRESHOLD) predict = sub.add_parser( - "predict", help="predict signed robust future-low q20/q80 from an exported 8H CSV" + "predict", help="predict signed robust future-low q20/q80 from an exported timeframe CSV" ) predict.add_argument("--artifact", type=Path, default=Path("artifacts/v1/model.joblib")) predict.add_argument("--csv", type=Path, required=True) @@ -74,7 +75,7 @@ def main() -> None: dsn=args.dsn, db_type=args.db_type, exchange=args.exchange, - end_ms=parse_time_ms(args.end_ms) if args.end_ms else None, + end_ms=parse_time_ms(args.end_ms, args.timeframe) if args.end_ms else None, history_months=args.history_months, holdout_months=args.holdout_months, min_age_days=args.min_age_days, @@ -82,6 +83,7 @@ def main() -> None: skip_download=args.skip_download, banbot_cmd=args.banbot_cmd, banbot_export_cmd=args.banbot_export_cmd, + timeframe=args.timeframe, ) print(json.dumps({"selected": len(result["symbols"]), "excluded": len(result["excluded"])}, indent=2)) elif args.command == "train": @@ -104,7 +106,7 @@ def main() -> None: dsn=args.dsn, db_type=args.db_type, exchange=args.exchange, - end_ms=parse_time_ms(args.end_ms) if args.end_ms else None, + end_ms=parse_time_ms(args.end_ms, args.timeframe) if args.end_ms else None, history_months=args.history_months, holdout_months=args.holdout_months, min_age_days=args.min_age_days, @@ -112,6 +114,7 @@ def main() -> None: skip_download=args.skip_download, banbot_cmd=args.banbot_cmd, banbot_export_cmd=args.banbot_export_cmd, + timeframe=args.timeframe, ) result = run_training( data_dir=args.out_dir, diff --git a/cbpred/experiments.py b/cbpred/experiments.py index d43d8ff..b21cede 100644 --- a/cbpred/experiments.py +++ b/cbpred/experiments.py @@ -31,13 +31,14 @@ from .pipeline import ( compute_indicators, evaluate_predictions, fit_asset_scaler, - floor_to_8h, + floor_to_timeframe, load_manifest, load_raw_frames, months_before, recent_split_boundaries, scale_asset_profile, split_holdout_symbols, + timeframe_step_ms, ) @@ -274,6 +275,7 @@ def build_variant_dataset( validation_end_ms: int, calibration_end_ms: int, volatility_rank_threshold: float, + step_ms: int = EIGHT_H_MS, ) -> tuple[pd.DataFrame, list[str], dict[str, dict[str, float]]]: indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()} raw_profiles = { @@ -292,7 +294,7 @@ def build_variant_dataset( train_set = set(train_symbols) validation_set = set(validation_symbols) test_set = set(test_symbols) - step = EIGHT_H_MS + step = int(step_ms) rows: list[pd.DataFrame] = [] candidate_counts = {"train": 0, "validation": 0, "calibration": 0, "test": 0} selected_counts = {name: 0 for name in candidate_counts} @@ -515,10 +517,13 @@ def _artifact( manifest: dict[str, Any], end_ms: int, test_start_ms: int, + timeframe: str, + step_ms: int, ) -> dict[str, Any]: return { "version": 4, - "timeframe": "8h", + "timeframe": timeframe, + "step_ms": step_ms, "window": WINDOW, "horizon": HORIZON, "target_kind": "signed_future_low_q20_return_atr_normalized", @@ -572,10 +577,12 @@ def run_experiments( ) -> dict[str, Any]: started = time.monotonic() output_dir.mkdir(parents=True, exist_ok=True) - _progress("loading manifest and raw 8H frames") + _progress("loading manifest and raw frames") manifest = load_manifest(data_dir) frames = load_raw_frames(data_dir, manifest) - end_ms = floor_to_8h(int(manifest["end_ms"])) + timeframe = str(manifest.get("timeframe", "8h")) + step_ms = int(manifest.get("step_ms") or timeframe_step_ms(timeframe)) + end_ms = floor_to_timeframe(int(manifest["end_ms"]), timeframe) test_start_ms = months_before(end_ms, test_months) indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()} raw_profiles = { @@ -585,13 +592,14 @@ def run_experiments( profiles = {symbol: profile for symbol, profile in raw_profiles.items() if profile is not None} train_symbols, validation_symbols, test_symbols = split_holdout_symbols(profiles, seed) scaler = fit_asset_scaler(profiles, train_symbols) - validation_end_ms, calibration_end_ms = recent_split_boundaries(test_start_ms, end_ms) + validation_end_ms, calibration_end_ms = recent_split_boundaries(test_start_ms, end_ms, step_ms) _progress( f"dataset={len(frames)} symbols; train/validation/test symbols=" f"{len(train_symbols)}/{len(validation_symbols)}/{len(test_symbols)}" ) stored_baseline = json.loads(baseline_report.read_text()) stored_baseline_score = _baseline_score(stored_baseline) + stored_test = stored_baseline.get("test") or stored_baseline.get("selected", {}).get("test", {}) _progress(f"stored baseline validation selection score={stored_baseline_score:.8f}") results: list[dict[str, Any]] = [] fitted_by_name: dict[str, tuple[FeatureVariant, dict[str, Any], pd.DataFrame, dict[str, dict[str, float]]]] = {} @@ -618,6 +626,7 @@ def run_experiments( validation_end_ms, calibration_end_ms, volatility_rank_threshold, + step_ms, ) filter_stats = dataset.attrs.get("volatility_filter", {}) _progress( @@ -673,7 +682,7 @@ def run_experiments( "validation_selection_score": comparison_baseline_score, "validation": next(item["validation"] for item in results if item["name"] == "baseline"), "test": baseline_test, - "stored_test": stored_baseline["test"], + "stored_test": stored_test, } best_test = _evaluate_test(best_dataset, best_fitted) best_dir = output_dir / "best" @@ -687,13 +696,22 @@ def run_experiments( manifest, end_ms, test_start_ms, + timeframe, + step_ms, ), best_dir / "model.joblib", ) report = { "protocol": { "data_dir": str(data_dir.resolve()), + "timeframe": timeframe, + "step_ms": step_ms, "end_ms": end_ms, + "history_start_ms": manifest.get("history_start_ms"), + "training_snapshot_ms": manifest.get("training_snapshot_ms"), + "min_age_days": manifest.get("min_age_days", 730), + "eligible_symbols": len(manifest.get("symbols", [])), + "excluded_symbols": len(manifest.get("excluded", [])), "test_start_ms": test_start_ms, "validation_end_ms": validation_end_ms, "calibration_end_ms": calibration_end_ms, @@ -724,9 +742,13 @@ def run_experiments( def _write_report_markdown(path: Path, report: dict[str, Any]) -> None: baseline = report["baseline"] selected = report["selected"] + protocol = report["protocol"] lines = [ "# Feature experiment report", "", + f"- timeframe: `{protocol.get('timeframe', '8h')}`", + f"- eligible/excluded symbols: `{protocol.get('eligible_symbols')}`/`{protocol.get('excluded_symbols')}`", + f"- min listing age: `{protocol.get('min_age_days', 730)}` days", f"- baseline: `{baseline['feature_count']}` features, validation score `{baseline['validation_selection_score']:.8f}`", f"- selected: `{selected['name']}`, `{selected['feature_count']}` features, validation score `{selected['validation_selection_score']:.8f}`", f"- selected test MAE/RMSE: `{selected['test']['mae_point']:.6f}` / `{selected['test']['rmse_point']:.6f}`", diff --git a/cbpred/latest_predict.py b/cbpred/latest_predict.py index d713b6e..40e9796 100644 --- a/cbpred/latest_predict.py +++ b/cbpred/latest_predict.py @@ -13,7 +13,7 @@ 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 +from .pipeline import load_database_url, parse_time_ms, predict_from_csv, prepare_data, timeframe_step_ms REPO_ROOT = Path(__file__).resolve().parents[1] @@ -42,6 +42,12 @@ def _database_endpoint(dsn: str) -> tuple[str, int]: return parsed.hostname or "127.0.0.1", parsed.port or 8812 +def output_columns_for_timeframe(timeframe: str) -> tuple[str, ...]: + timeframe_step_ms(timeframe) + volume_column = f"quote_volume_usdt_{timeframe}" + return (*OUTPUT_COLUMNS[:4], volume_column, *OUTPUT_COLUMNS[5:]) + + def ensure_database( banbot_config: Path, dsn: str | None, @@ -86,6 +92,9 @@ 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) + timeframe = str(manifest.get("timeframe", artifact_data.get("timeframe", "8h"))) + output_columns = output_columns_for_timeframe(timeframe) + volume_column = output_columns[4] trained_symbols = set(artifact_data.get("asset_profiles", {})) if trained_symbols: skipped_symbols = [ @@ -117,7 +126,7 @@ def build_predictions(data_dir: Path, artifact: Path, output_dir: Path) -> dict[ "symbol": result["symbol"], "latest_bar_start_utc": _utc_label(int(result["timestamp"])), "close": result["close"], - "quote_volume_usdt_8h": float(latest["quote"]), + volume_column: float(latest["quote"]), "atr_rank_200_pct": ( result["volatility_rank_200"] * 100 if result["volatility_rank_200"] is not None @@ -134,19 +143,20 @@ def build_predictions(data_dir: Path, artifact: Path, output_dir: Path) -> dict[ if index == 1 or index == total or index % 25 == 0: print(f"[latest-predict] predicted {index}/{total}", flush=True) - rows.sort(key=lambda row: float(row["quote_volume_usdt_8h"]), reverse=True) + rows.sort(key=lambda row: float(row[volume_column]), reverse=True) for rank, row in enumerate(rows, 1): row["rank"] = rank output_dir.mkdir(parents=True, exist_ok=True) output_csv = output_dir / "predictions.csv" with output_csv.open("w", newline="") as handle: - writer = csv.DictWriter(handle, fieldnames=OUTPUT_COLUMNS) + writer = csv.DictWriter(handle, fieldnames=output_columns) writer.writeheader() writer.writerows(rows) summary = { "model": str(artifact.resolve()), + "timeframe": timeframe, "symbols": len(rows), "skipped_symbols": skipped_symbols, "eligible_predictions": sum(row["drawdown_point_pct"] is not None for row in rows), @@ -176,6 +186,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--dsn", default=None) parser.add_argument("--db-type", choices=["auto", "questdb", "timescale"], default="auto") parser.add_argument("--exchange", default="binance") + parser.add_argument("--timeframe", choices=["8h", "1d"], default="8h") return parser @@ -192,11 +203,12 @@ def main() -> None: dsn=args.dsn, db_type=args.db_type, exchange=args.exchange, - end_ms=parse_time_ms(args.end) if args.end else None, + end_ms=parse_time_ms(args.end, args.timeframe) if args.end else None, history_months=args.history_months, holdout_months=args.holdout_months, min_age_days=args.min_age_days, chunk_size=args.chunk_size, + timeframe=args.timeframe, ) print( f"[latest-predict] exported {len(manifest['symbols'])} symbols; " diff --git a/cbpred/pipeline.py b/cbpred/pipeline.py index d5a7d0c..c6b629a 100644 --- a/cbpred/pipeline.py +++ b/cbpred/pipeline.py @@ -25,7 +25,12 @@ from sklearn.metrics import mean_absolute_error, mean_squared_error EIGHT_H_MS = 8 * 60 * 60 * 1000 +DAY_MS = 24 * 60 * 60 * 1000 HOUR_MS = 60 * 60 * 1000 +TIMEFRAME_STEP_MS = { + "8h": EIGHT_H_MS, + "1d": DAY_MS, +} WINDOW = 18 HORIZON = 6 VOLATILITY_RANK_WINDOW = 200 @@ -62,9 +67,21 @@ class Symbol: combined: bool = False -def parse_time_ms(value: str | int | None) -> int: +def timeframe_step_ms(timeframe: str) -> int: + try: + return TIMEFRAME_STEP_MS[str(timeframe).lower()] + except KeyError as exc: + raise ValueError(f"unsupported timeframe: {timeframe}; expected 8h or 1d") from exc + + +def floor_to_timeframe(value_ms: int, timeframe: str) -> int: + step_ms = timeframe_step_ms(timeframe) + return value_ms - value_ms % step_ms + + +def parse_time_ms(value: str | int | None, timeframe: str = "8h") -> int: if value is None: - return floor_to_8h(int(datetime.now(timezone.utc).timestamp() * 1000)) + return floor_to_timeframe(int(datetime.now(timezone.utc).timestamp() * 1000), timeframe) if isinstance(value, int): return value text = str(value).strip() @@ -83,7 +100,7 @@ def parse_time_ms(value: str | int | None) -> int: def floor_to_8h(value_ms: int) -> int: - return value_ms - value_ms % EIGHT_H_MS + return floor_to_timeframe(value_ms, "8h") def months_before(value_ms: int, months: int) -> int: @@ -237,6 +254,7 @@ def run_banbot_export( history_start_ms: int, end_ms: int, banbot_export_cmd: str = "go run ./cmd/cbpred-export", + timeframe: str = "8h", ) -> None: if not symbols: raise ValueError("no eligible symbols to export") @@ -253,6 +271,8 @@ def run_banbot_export( str(history_start_ms), "--end-ms", str(end_ms), + "--timeframe", + timeframe, "--out-dir", str(out_dir), ] @@ -260,7 +280,9 @@ def run_banbot_export( subprocess.run(command, cwd=project_dir, check=True) -def aggregate_8h(rows: Iterable[tuple[Any, ...]]) -> pd.DataFrame: +def aggregate_ohlcv(rows: Iterable[tuple[Any, ...]], timeframe: str = "8h") -> pd.DataFrame: + step_ms = timeframe_step_ms(timeframe) + bars_per_bucket = step_ms // HOUR_MS columns = CSV_COLUMNS values = list(rows) if not values: @@ -275,12 +297,14 @@ def aggregate_8h(rows: Iterable[tuple[Any, ...]]) -> pd.DataFrame: frame["buy_volume"] = frame["buy_volume"].fillna(0.0) frame["trade_num"] = frame["trade_num"].fillna(0).astype(np.int64) frame["quote"] = np.where(frame["quote"] > 0, frame["quote"], frame["volume"] * frame["close"]) - frame["bucket"] = (frame["timestamp"] // EIGHT_H_MS) * EIGHT_H_MS + frame["bucket"] = (frame["timestamp"] // step_ms) * step_ms output: list[dict[str, Any]] = [] for bucket, group in frame.groupby("bucket", sort=True): group = group.sort_values("timestamp") - expected = bucket + np.arange(8, dtype=np.int64) * HOUR_MS - if len(group) != 8 or not np.array_equal(group["timestamp"].to_numpy(), expected): + expected = bucket + np.arange(bars_per_bucket, dtype=np.int64) * HOUR_MS + if len(group) != bars_per_bucket or not np.array_equal( + group["timestamp"].to_numpy(), expected + ): continue output.append( { @@ -298,6 +322,10 @@ def aggregate_8h(rows: Iterable[tuple[Any, ...]]) -> pd.DataFrame: return pd.DataFrame(output, columns=columns) +def aggregate_8h(rows: Iterable[tuple[Any, ...]]) -> pd.DataFrame: + return aggregate_ohlcv(rows, "8h") + + def symbol_filename(symbol: Symbol) -> str: slug = symbol.symbol.replace("/", "_").replace(":", "_") return f"{symbol.sid}_{slug}.csv" @@ -318,11 +346,13 @@ def prepare_data( skip_download: bool = False, banbot_cmd: str = "go run .", banbot_export_cmd: str = "go run ./cmd/cbpred-export", + timeframe: str = "8h", ) -> dict[str, Any]: + step_ms = timeframe_step_ms(timeframe) out_dir.mkdir(parents=True, exist_ok=True) raw_dir = out_dir / "raw" raw_dir.mkdir(parents=True, exist_ok=True) - end_ms = floor_to_8h(end_ms or parse_time_ms(None)) + end_ms = floor_to_timeframe(end_ms or parse_time_ms(None, timeframe), timeframe) history_start_ms = months_before(end_ms, history_months) training_snapshot_ms = months_before(end_ms, holdout_months) database_url = load_database_url(banbot_config, dsn) @@ -356,12 +386,13 @@ def prepare_data( history_start_ms, end_ms, banbot_export_cmd, + timeframe, ) manifest_symbols: list[dict[str, Any]] = [] for symbol in symbols: path = raw_dir / symbol_filename(symbol) if not path.exists(): - excluded.append({**symbol.__dict__, "reason": "no_complete_8h_bars"}) + excluded.append({**symbol.__dict__, "reason": f"no_complete_{timeframe}_bars"}) continue frame = pd.read_csv(path) frame["timestamp"] = pd.to_numeric(frame["timestamp"], errors="coerce") @@ -370,18 +401,21 @@ def prepare_data( frame = frame[frame["timestamp"] < symbol.delist_ms] frame.to_csv(path, index=False, float_format="%.12g") if frame.empty: - excluded.append({**symbol.__dict__, "reason": "no_complete_8h_bars"}) + excluded.append({**symbol.__dict__, "reason": f"no_complete_{timeframe}_bars"}) path.unlink(missing_ok=True) continue manifest_symbols.append({**symbol.__dict__, "path": str(path.relative_to(out_dir)), "bars": len(frame)}) manifest = { "version": 2, - "timeframe": "8h", + "timeframe": timeframe, "storage_timeframe": "banbot_orm", - "step_ms": EIGHT_H_MS, + "step_ms": step_ms, "end_ms": end_ms, "history_start_ms": history_start_ms, "training_snapshot_ms": training_snapshot_ms, + "history_months": history_months, + "holdout_months": holdout_months, + "min_age_days": min_age_days, "db_type": actual_db_type, "exchange": exchange, "symbols": manifest_symbols, @@ -727,6 +761,7 @@ def build_feature_dataset( validation_end_ms: int | None = None, calibration_end_ms: int | None = None, volatility_rank_threshold: float = VOLATILITY_RANK_THRESHOLD, + step_ms: int = EIGHT_H_MS, ) -> tuple[pd.DataFrame, list[str], dict[str, dict[str, float]]]: if not 0.0 <= volatility_rank_threshold <= 1.0: raise ValueError("volatility_rank_threshold must be between 0 and 1") @@ -745,7 +780,7 @@ def build_feature_dataset( rows: list[pd.DataFrame] = [] candidate_counts = {"train": 0, "validation": 0, "calibration": 0, "test": 0, "recent": 0} selected_counts = {name: 0 for name in candidate_counts} - step = EIGHT_H_MS + step = int(step_ms) explicit_splits = train_symbols is not None and validation_symbols is not None and test_symbols is not None train_set = set(train_symbols or []) validation_set = set(validation_symbols or []) @@ -874,13 +909,15 @@ def split_holdout_symbols(symbols: Iterable[str], seed: int = 42, holdout_ratio: return train_symbols, validation, test -def recent_split_boundaries(start_ms: int, end_ms: int) -> tuple[int, int]: +def recent_split_boundaries( + start_ms: int, end_ms: int, step_ms: int = EIGHT_H_MS +) -> tuple[int, int]: """Use the first 30% for model selection, next 20% for calibration, last 50% for final test.""" duration = end_ms - start_ms - validation_end = floor_to_8h(start_ms + int(duration * 0.30)) - calibration_end = floor_to_8h(start_ms + int(duration * 0.50)) - validation_end = max(start_ms + EIGHT_H_MS, min(validation_end, end_ms - 2 * EIGHT_H_MS)) - calibration_end = max(validation_end + EIGHT_H_MS, min(calibration_end, end_ms - EIGHT_H_MS)) + validation_end = start_ms + (int(duration * 0.30) // step_ms) * step_ms + calibration_end = start_ms + (int(duration * 0.50) // step_ms) * step_ms + validation_end = max(start_ms + step_ms, min(validation_end, end_ms - 2 * step_ms)) + calibration_end = max(validation_end + step_ms, min(calibration_end, end_ms - step_ms)) return validation_end, calibration_end @@ -1182,12 +1219,14 @@ def run_training( ) -> dict[str, Any]: if not 0.0 <= volatility_rank_threshold <= 1.0: raise ValueError("volatility_rank_threshold must be between 0 and 1") - _progress("training: loading manifest and raw 8H frames") + _progress("training: loading manifest and raw frames") manifest = load_manifest(data_dir) frames = load_raw_frames(data_dir, manifest) if len(frames) < 3: raise ValueError("at least 3 symbols with raw data are required") - end_ms = floor_to_8h(end_ms or int(manifest["end_ms"])) + timeframe = str(manifest.get("timeframe", "8h")) + step_ms = int(manifest.get("step_ms") or timeframe_step_ms(timeframe)) + end_ms = floor_to_timeframe(end_ms or int(manifest["end_ms"]), timeframe) test_start_ms = months_before(end_ms, test_months) profile_indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()} raw_profiles = { @@ -1199,7 +1238,7 @@ def run_training( raise ValueError("at least 3 symbols with enough pre-test history are required") train_symbols, validation_symbols, test_symbols = split_holdout_symbols(profiles, seed) asset_scaler = fit_asset_scaler(profiles, train_symbols) - validation_end_ms, calibration_end_ms = recent_split_boundaries(test_start_ms, end_ms) + validation_end_ms, calibration_end_ms = recent_split_boundaries(test_start_ms, end_ms, step_ms) _progress( f"training: {len(frames)} symbols, {len(profiles)} profiles; " f"train/validation/test symbols={len(train_symbols)}/{len(validation_symbols)}/{len(test_symbols)}" @@ -1226,6 +1265,7 @@ def run_training( validation_end_ms, calibration_end_ms, volatility_rank_threshold, + step_ms, ) filter_stats = dataset.attrs.get("volatility_filter", {}) _progress( @@ -1367,7 +1407,8 @@ def run_training( artifact_dir.mkdir(parents=True, exist_ok=True) artifact = { "version": 3, - "timeframe": "8h", + "timeframe": timeframe, + "step_ms": step_ms, "window": WINDOW, "horizon": HORIZON, "target_kind": TARGET_KIND, @@ -1396,6 +1437,15 @@ def run_training( joblib.dump(artifact, artifact_dir / "model.joblib") report = { "protocol": { + "timeframe": timeframe, + "step_ms": step_ms, + "data_dir": str(data_dir.resolve()), + "end_ms": end_ms, + "history_start_ms": manifest.get("history_start_ms"), + "training_snapshot_ms": manifest.get("training_snapshot_ms"), + "min_age_days": manifest.get("min_age_days", 730), + "eligible_symbols": len(manifest.get("symbols", [])), + "excluded_symbols": len(manifest.get("excluded", [])), "train_anchor_before": test_start_ms, "recent_window_start": test_start_ms, "recent_window_end": end_ms, @@ -1485,6 +1535,8 @@ def predict_from_csv( raise ValueError( "artifact uses an unsupported target or requires full-universe feature context" ) + timeframe = str(artifact.get("timeframe", "8h")) + step_ms = int(artifact.get("step_ms") or timeframe_step_ms(timeframe)) frame = pd.read_csv(csv_path) for column in CSV_COLUMNS: frame[column] = pd.to_numeric(frame[column], errors="coerce") @@ -1497,7 +1549,7 @@ def predict_from_csv( raise ValueError("--at is earlier than the model training snapshot") matches = np.flatnonzero(frame["timestamp"].to_numpy(dtype=np.int64) == int(at_ms)) if len(matches) == 0: - raise ValueError("--at must match an 8H bar timestamp in the CSV") + raise ValueError(f"--at must match a {timeframe} bar timestamp in the CSV") anchor_index = int(matches[-1]) if symbol is None: for candidate, path in artifact.get("symbol_files", {}).items(): @@ -1519,6 +1571,8 @@ def predict_from_csv( ) response = { "timestamp": int(frame.iloc[anchor_index]["timestamp"]), + "timeframe": timeframe, + "step_ms": step_ms, "close": float(frame.iloc[anchor_index]["close"]), "model": artifact["model_name"], "feature_stage": feature_stage, diff --git a/cmd/cbpred-export/main.go b/cmd/cbpred-export/main.go index 91a5a57..63de2c0 100644 --- a/cmd/cbpred-export/main.go +++ b/cmd/cbpred-export/main.go @@ -25,21 +25,22 @@ func main() { pairs := flag.String("pairs", "", "comma-separated symbols") startMS := flag.Int64("start-ms", 0, "inclusive start timestamp in milliseconds") endMS := flag.Int64("end-ms", 0, "exclusive end timestamp in milliseconds") + timeframe := flag.String("timeframe", "8h", "bar timeframe (8h or 1d)") outDir := flag.String("out-dir", "", "CSV output directory") flag.Parse() - if *configPath == "" || *dataDir == "" || *pairs == "" || *outDir == "" || *startMS <= 0 || *endMS <= *startMS { + if (*timeframe != "8h" && *timeframe != "1d") || *configPath == "" || *dataDir == "" || *pairs == "" || *outDir == "" || *startMS <= 0 || *endMS <= *startMS { flag.Usage() os.Exit(2) } args := &config.CmdArgs{ - Configs: config.ArrString{*configPath}, - DataDir: *dataDir, - NoDefault: true, - RawPairs: *pairs, - TimeStart: strconv.FormatInt(*startMS, 10), - TimeEnd: strconv.FormatInt(*endMS, 10), - LogLevel: "error", + Configs: config.ArrString{*configPath}, + DataDir: *dataDir, + NoDefault: true, + RawPairs: *pairs, + TimeStart: strconv.FormatInt(*startMS, 10), + TimeEnd: strconv.FormatInt(*endMS, 10), + LogLevel: "error", } if err := biz.SetupComs(args); err != nil { fail(err) @@ -58,7 +59,7 @@ func main() { if err != nil { fail(err) } - _, rows, err := orm.GetSeries(exs, "8h", *startMS, *endMS, 0, false) + _, rows, err := orm.GetSeries(exs, *timeframe, *startMS, *endMS, 0, false) if err != nil { fail(err) } diff --git a/docs/1d_2026-08-24.md b/docs/1d_2026-08-24.md new file mode 100644 index 0000000..d4be973 --- /dev/null +++ b/docs/1d_2026-08-24.md @@ -0,0 +1,77 @@ +# 1d 模型实验报告 + +## 结论 + +在与当前 8H 正式模型相同的模型、特征、过滤和切分协议下，1d 版本成功完成全市场数据导出、训练、验证、最终测试和最新预测。验证集仍选择 `compact6`，模型使用 146 个特征、ExtraTrees 点模型和两个 HistGradientBoosting 分位数模型。 + +1d 最终测试点预测 MAE 为 `0.043335`，RMSE 为 `0.056943`。q80 覆盖率为 `72.76%`，q20 下界命中率为 `93.68%`。 + +## 数据快照 + +| 项目 | 值 | +| --- | --- | +| 周期 | 1d | +| 快照结束 | 2026-08-24 00:00 UTC | +| 历史范围 | 36 个月 | +| 历史起点 | 2023-08-24 00:00 UTC | +| 训练快照 | 2025-10-24 00:00 UTC | +| 最低上市年龄 | 730 天 | +| 市场过滤 | Binance linear、USDT 结算、非组合、未在训练快照前退市 | +| 保留/排除品种 | 151 / 2,125 | +| 原始行情文件 | 151 个 CSV，约 15 MB | +| bar 连续性检查 | 全部文件相邻时间差均为 24h | + +原始数据和筛选清单：[`data_1d/manifest.json`](../data_1d/manifest.json) 与 [`data_1d/raw/`](../data_1d/raw/)。 + +## 训练协议 + +- 输入最近 18 根已收盘 1d K 线。 +- 预测未来 6 根 1d K 线的最低价 20% 分位数收益。 +- 目标按品种训练期 `ATR/close` 高波动 20% 样本中位数归一化。 +- 训练/验证/测试品种数量为 `90/30/31`。 +- 最近 10 个月按时间切成验证 30%、校准 20%、最终测试 50%。 +- 训练/验证/校准/最终测试样本数为 `68,840/2,550/1,650/4,526`。 +- `atr_rank_200` 窗口为 200 根日线，样本过滤阈值为 `0.0`，因此不主动排除低波动样本。 +- 随机种子为 `42`，点模型树数为 `20`，与当前正式 8H 实验配置一致。 + +## 特征选择 + +| 版本 | 特征数 | 验证选择分数 | 测试 MAE | 测试 RMSE | +| --- | ---: | ---: | ---: | ---: | +| baseline | 182 | 0.17235337 | 0.043468 | 0.057143 | +| compact6 | 146 | 0.17234073 | 0.043335 | 0.056943 | + +`compact6` 在验证集胜出，因此作为 1d 产物保存。它相比 baseline 的测试 MAE 下降约 `0.31%`，RMSE 下降约 `0.35%`；测试集只用于最终报告，没有参与选择。 + +## 最终测试指标 + +| 指标 | compact6 | +| --- | ---: | +| 样本数 | 4,526 | +| 点预测 MAE | 0.043335 | +| 点预测 RMSE | 0.056943 | +| q20 pinball | 0.021642 | +| q80 pinball | 0.011617 | +| q20 coverage | 6.32% | +| q80 coverage | 72.76% | +| q20 下界命中率 | 93.68% | + +误差指标在乘回品种 ATR 尺度后的实际收益空间计算。日线未来 6 根约对应 6 天，8H 未来 6 根约对应 2 天，因此 1d 和 8H 的绝对误差不能简单解释为同一预测期限的优劣。 + +## 最新批量预测 + +基于同一快照最后一根完整日线 `2026-08-23 00:00 UTC`，151/151 个品种满足预测资格。按 1d USDT 成交额排序的前五个品种为： + +| 排名 | 品种 | 1d 成交额 USDT | 点预测回撤 | +| ---: | --- | ---: | ---: | +| 1 | ETH/USDT:USDT | 10,761,726,205 | 4.14% | +| 2 | BTC/USDT:USDT | 10,079,189,363 | 2.46% | +| 3 | SOL/USDT:USDT | 2,411,667,199 | 5.02% | +| 4 | ZEC/USDT:USDT | 2,306,432,737 | 6.27% | +| 5 | XRP/USDT:USDT | 2,174,780,164 | 4.51% | + +预测 CSV 和摘要：[`predictions/latest_1d/predictions.csv`](../predictions/latest_1d/predictions.csv) 与 [`predictions/latest_1d/report.json`](../predictions/latest_1d/report.json)。模型文件：[`artifacts_1d/best/model.joblib`](../artifacts_1d/best/model.joblib)。 + +## 限制 + +这是同一历史快照上的离线实验，不能作为交易信号或未来收益保证。日线的样本数量明显少于 8H，且 6 根日线覆盖更长的未来区间；q80 覆盖率低于名义 80%，说明当前校准在这份 1d 测试集上仍有偏差。banbot 导出过程中有两个既有 1d pending job 的可见性告警，但本次导出命令成功完成，最终文件连续性检查全部通过。 diff --git a/docs/model.md b/docs/model.md index 4bccb4a..98ead96 100644 --- a/docs/model.md +++ b/docs/model.md @@ -123,3 +123,11 @@ price_level = close × (1 + future_low_return) - 模型只对保存了训练画像且有足够历史的品种预测。当前 ATR 排名至少需要 200 根 8H K 线。 - `atr_rank_200_pct` 是状态描述，不代表模型只预测高波动时刻；正式阈值为 `0.0`。 - 预测结果中的空值表示当前时刻不满足模型资格条件，不应替换成零。 + +## 1d 实验结果 + +1d 实验使用与默认模型相同的 `compact6` 特征（146 个特征）、ExtraTrees 点模型、两个 HistGradientBoosting 分位数模型、seed=42、20 棵树、36 个月数据、最后 10 个月留出、上市至少 730 天和 ATR 排名阈值 0.0。仅将 bar 周期、连续性步长和对应的 18/6 根窗口改为 1d。 + +固定快照截止 `2026-08-24 00:00 UTC`，导出 151 个品种、排除 2,125 个品种。品种按 90/30/31 拆分为训练/验证/测试，样本按训练/验证/校准/最终测试为 68,840/2,550/1,650/4,526。验证选择分数为 `0.17234073`，1d 测试点预测 MAE/RMSE 为 `0.043335/0.056943`，q80 覆盖率为 `72.76%`，q20 下界命中率为 `93.68%`。 + +详细机器可读报告和模型分别是 [`artifacts_1d/report.json`](../artifacts_1d/report.json)、[`artifacts_1d/report.md`](../artifacts_1d/report.md) 和 [`artifacts_1d/best/model.joblib`](../artifacts_1d/best/model.joblib)。这些指标在真实收益空间中计算，日线的 6 根预测跨度与 8H 的 6 根预测跨度不同，不应直接当作同一持有期的收益比较。 diff --git a/docs/usage.md b/docs/usage.md index ae317cf..a917c85 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -31,7 +31,7 @@ python3 -m venv .venv 1. 检查配置中的 QuestDB 是否可连接；QuestDB 默认不可用时，通过 banbot 的导出命令触发其启动流程。 2. 从数据库筛选 Binance `linear` 市场、结算币为 USDT、上市至少 730 天且未退市的品种。 3. 调用 banbot 下载最近 36 个月的 1H K 线。 -4. 调用 banbot 导出原始 K 线，并只保留连续完整的 8 根 1H K 线聚合成 8H K 线。 +4. 调用 banbot 导出请求周期的完整 K 线；默认是 8H，也可以通过 `--timeframe 1d` 导出日线。 5. 写入 `data/manifest.json` 和 `data/raw/*.csv`。 6. 加载 `artifacts/v1/model.joblib`，对清单中的每个品种使用最新完整 8H K 线预测。 7. 按最新 8H USDT 成交额降序写入 `predictions/latest_v1/predictions.csv`，并写入 `report.json`。 @@ -48,7 +48,7 @@ python3 -m venv .venv --end 2026-08-23T00:00:00Z ``` -`--end` 用于固定数据快照，支持 UTC 日期、ISO 时间或毫秒时间戳。默认使用当前 UTC 时间向下取整到 8H 边界。 +`--end` 用于固定数据快照，支持 UTC 日期、ISO 时间或毫秒时间戳。默认按 `--timeframe` 向下取整到对应边界。 ## 手动命令 @@ -71,6 +71,15 @@ python3 -m venv .venv `prepare` 的默认范围是最近 36 个月，训练快照前的品种至少上市 730 天，最近 10 个月预留给测试。`manifest.json` 记录筛选出的品种、数据范围、数据库类型和被排除的原因。 +导出日线时使用独立目录，避免覆盖 8H 快照： + +```bash +.venv/bin/python -m cbpred prepare \ + --timeframe 1d \ + --out-dir data_1d \ + --banbot-config /home/ctyun/ban/banbot/biz/config.yml +``` + ### 训练模型 ```bash @@ -96,9 +105,9 @@ python3 -m venv .venv --symbol BTC/USDT:USDT ``` -不传 `--at` 时使用 CSV 中最新完整 8H K 线。传入 `--at` 时，时间必须精确匹配一个 8H K 线起始时间，且不能早于模型训练快照。 +不传 `--at` 时使用 CSV 中最新完整周期 K 线。传入 `--at` 时，时间必须精确匹配一个模型周期 K 线起始时间，且不能早于模型训练快照。 -模型需要至少 200 根 8H K 线计算 ATR 排名，并需要模型中保存的该品种训练画像。`eligible_for_model` 为 `false` 时，预测数值返回空值，并包含 `skip_reason`。 +模型需要至少 200 根对应周期 K 线计算 ATR 排名，并需要模型中保存的该品种训练画像。`eligible_for_model` 为 `false` 时，预测数值返回空值，并包含 `skip_reason`。 ## 文件布局 @@ -123,7 +132,7 @@ cbpred/ - QuestDB 连接失败：检查 banbot 配置中的数据库地址、QuestDB 服务和端口；一键脚本只对默认 QuestDB 端口尝试触发 banbot 启动。 - `missing data/manifest.json`：先运行 `prepare`，或确认一键脚本没有在下载/导出阶段失败。 - 某品种预测为空：检查 `eligible_for_model` 和 `skip_reason`，通常是 200 根 ATR 历史不足或该品种没有正式模型画像。 -- 数据只有部分 8H K 线：导出会丢弃不连续或不完整的 8 根 1H 桶，这是为了避免特征和未来标签跨越缺口。 +- 数据只有部分 K 线：导出和训练会丢弃不连续或不完整的 bar，这是为了避免特征和未来标签跨越缺口。 ## 验证 diff --git a/tests/test_artifact_paths.py b/tests/test_artifact_paths.py index 9c81383..8d7edf7 100644 --- a/tests/test_artifact_paths.py +++ b/tests/test_artifact_paths.py @@ -15,6 +15,8 @@ class ArtifactPathTests(unittest.TestCase): self.assertEqual(train.artifact_dir, Path("artifacts/v1")) self.assertEqual(run.artifact_dir, Path("artifacts/v1")) self.assertEqual(predict.artifact, Path("artifacts/v1/model.joblib")) + daily = parser.parse_args(["prepare", "--timeframe", "1d"]) + self.assertEqual(daily.timeframe, "1d") if __name__ == "__main__": diff --git a/tests/test_latest_predict.py b/tests/test_latest_predict.py index 8e496ae..51df24e 100644 --- a/tests/test_latest_predict.py +++ b/tests/test_latest_predict.py @@ -6,7 +6,7 @@ from unittest.mock import patch import pandas as pd -from cbpred.latest_predict import OUTPUT_COLUMNS, build_predictions +from cbpred.latest_predict import OUTPUT_COLUMNS, build_predictions, output_columns_for_timeframe class LatestPredictTests(unittest.TestCase): @@ -27,6 +27,9 @@ class LatestPredictTests(unittest.TestCase): ), ) + def test_output_columns_follow_timeframe(self): + self.assertEqual(output_columns_for_timeframe("1d")[4], "quote_volume_usdt_1d") + def test_batch_skips_symbols_without_training_profiles(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 2f38c58..b647256 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -10,7 +10,9 @@ import pandas as pd from cbpred.pipeline import ( CSV_COLUMNS, + DAY_MS, EIGHT_H_MS, + aggregate_ohlcv, aggregate_8h, build_feature_dataset, compute_indicators, @@ -45,6 +47,33 @@ class PipelineTests(unittest.TestCase): self.assertEqual(frame.iloc[0]["timestamp"], 0) self.assertAlmostEqual(frame.iloc[0]["quote"], sum(row[6] for row in rows[:8])) + def test_aggregate_1d_drops_incomplete_buckets(self): + rows = make_hour_rows(0, 49) + frame = aggregate_ohlcv(rows, "1d") + self.assertEqual(len(frame), 2) + self.assertEqual(frame.iloc[0]["timestamp"], 0) + self.assertEqual(frame.iloc[1]["timestamp"], DAY_MS) + + def test_daily_feature_dataset_uses_daily_continuity(self): + raw = aggregate_ohlcv(make_hour_rows(0, 24 * 80), "1d") + profiles = { + "x": { + "asset_atr_price_top20_median": 0.01, + "asset_quote_volume_top20_median_log": 12.0, + } + } + scaler = fit_asset_scaler(profiles) + dataset, _, _ = build_feature_dataset( + {"x": raw}, + 1, + int(raw.iloc[55]["timestamp"]), + int(raw.iloc[-1]["timestamp"] + DAY_MS), + scaler, + volatility_rank_threshold=0.0, + step_ms=DAY_MS, + ) + self.assertFalse(dataset.empty) + def test_future_target_changes_but_past_features_do_not(self): start = 0 raw = aggregate_8h(make_hour_rows(start, 8 * 70))