支持1d级别插针模型

This commit is contained in:
2026-08-25 09:23:30 +08:00
parent baddd7763d
commit c0df80154b
12 changed files with 299 additions and 65 deletions

View File

@@ -1,15 +1,11 @@
# cbpred # 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。默认路径是: 运行环境需要 Python 3、Go、可访问的 banbot 源码和 banbot 配置中的 QuestDB。默认路径是:
@@ -81,6 +77,24 @@ predictions/latest_v1/report.json
如果数据已经由 banbot 更新完成,可在 `prepare` 中加入 `--skip-download`,只执行筛选、导出和整理。 如果数据已经由 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
一键脚本生成的 CSV 固定只包含以下 10 列: 一键脚本生成的 CSV 固定只包含以下 10 列:
@@ -88,11 +102,11 @@ predictions/latest_v1/report.json
| 列 | 含义 | | 列 | 含义 |
| --------------------------- | ------------------------------------- | | --------------------------- | ------------------------------------- |
| `rank` | 按 8H USDT 成交额从高到低的排名 | | `rank` | 按请求周期 USDT 成交额从高到低的排名 |
| `symbol` | 合约符号 | | `symbol` | 合约符号 |
| `latest_bar_start_utc` | 最新完整 8H K 线的 UTC 起始时间 | | `latest_bar_start_utc` | 最新完整请求周期 K 线的 UTC 起始时间 |
| `close` | 最新完整 8H K 线收盘价 | | `close` | 最新完整请求周期 K 线收盘价 |
| `quote_volume_usdt_8h` | 最新 8H 的 USDT 成交额 | | `quote_volume_usdt_8h` / `quote_volume_usdt_1d` | 最新请求周期的 USDT 成交额 |
| `atr_rank_200_pct` | 当前 ATR 在过去 200 根 8H ATR 中的百分位,单位为 `%` | | `atr_rank_200_pct` | 当前 ATR 在过去 200 根 8H ATR 中的百分位,单位为 `%` |
| `drawdown_point_pct` | 点预测对应的回撤百分比 | | `drawdown_point_pct` | 点预测对应的回撤百分比 |
| `drawdown_cap_80pct_pct` | q80 对应的回撤百分比上界参考 | | `drawdown_cap_80pct_pct` | q80 对应的回撤百分比上界参考 |

View File

@@ -14,16 +14,17 @@ from .pipeline import (
def build_parser() -> argparse.ArgumentParser: 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) 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("--out-dir", type=Path, default=Path("data"))
prepare.add_argument("--banbot-dir", type=Path, default=Path("/home/ctyun/ban/banbot")) 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("--banbot-config", type=Path, default=Path("/home/ctyun/ban/banbot/biz/config.yml"))
prepare.add_argument("--dsn", default=None) prepare.add_argument("--dsn", default=None)
prepare.add_argument("--db-type", choices=["auto", "questdb", "timescale"], default="auto") prepare.add_argument("--db-type", choices=["auto", "questdb", "timescale"], default="auto")
prepare.add_argument("--exchange", default="binance") 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("--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("--history-months", type=int, default=36)
prepare.add_argument("--holdout-months", type=int, default=10) 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) run.add_argument("--volatility-rank-threshold", type=float, default=VOLATILITY_RANK_THRESHOLD)
predict = sub.add_parser( 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("--artifact", type=Path, default=Path("artifacts/v1/model.joblib"))
predict.add_argument("--csv", type=Path, required=True) predict.add_argument("--csv", type=Path, required=True)
@@ -74,7 +75,7 @@ def main() -> None:
dsn=args.dsn, dsn=args.dsn,
db_type=args.db_type, db_type=args.db_type,
exchange=args.exchange, 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, history_months=args.history_months,
holdout_months=args.holdout_months, holdout_months=args.holdout_months,
min_age_days=args.min_age_days, min_age_days=args.min_age_days,
@@ -82,6 +83,7 @@ def main() -> None:
skip_download=args.skip_download, skip_download=args.skip_download,
banbot_cmd=args.banbot_cmd, banbot_cmd=args.banbot_cmd,
banbot_export_cmd=args.banbot_export_cmd, banbot_export_cmd=args.banbot_export_cmd,
timeframe=args.timeframe,
) )
print(json.dumps({"selected": len(result["symbols"]), "excluded": len(result["excluded"])}, indent=2)) print(json.dumps({"selected": len(result["symbols"]), "excluded": len(result["excluded"])}, indent=2))
elif args.command == "train": elif args.command == "train":
@@ -104,7 +106,7 @@ def main() -> None:
dsn=args.dsn, dsn=args.dsn,
db_type=args.db_type, db_type=args.db_type,
exchange=args.exchange, 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, history_months=args.history_months,
holdout_months=args.holdout_months, holdout_months=args.holdout_months,
min_age_days=args.min_age_days, min_age_days=args.min_age_days,
@@ -112,6 +114,7 @@ def main() -> None:
skip_download=args.skip_download, skip_download=args.skip_download,
banbot_cmd=args.banbot_cmd, banbot_cmd=args.banbot_cmd,
banbot_export_cmd=args.banbot_export_cmd, banbot_export_cmd=args.banbot_export_cmd,
timeframe=args.timeframe,
) )
result = run_training( result = run_training(
data_dir=args.out_dir, data_dir=args.out_dir,

View File

@@ -31,13 +31,14 @@ from .pipeline import (
compute_indicators, compute_indicators,
evaluate_predictions, evaluate_predictions,
fit_asset_scaler, fit_asset_scaler,
floor_to_8h, floor_to_timeframe,
load_manifest, load_manifest,
load_raw_frames, load_raw_frames,
months_before, months_before,
recent_split_boundaries, recent_split_boundaries,
scale_asset_profile, scale_asset_profile,
split_holdout_symbols, split_holdout_symbols,
timeframe_step_ms,
) )
@@ -274,6 +275,7 @@ def build_variant_dataset(
validation_end_ms: int, validation_end_ms: int,
calibration_end_ms: int, calibration_end_ms: int,
volatility_rank_threshold: float, volatility_rank_threshold: float,
step_ms: int = EIGHT_H_MS,
) -> tuple[pd.DataFrame, list[str], dict[str, dict[str, float]]]: ) -> tuple[pd.DataFrame, list[str], dict[str, dict[str, float]]]:
indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()} indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()}
raw_profiles = { raw_profiles = {
@@ -292,7 +294,7 @@ def build_variant_dataset(
train_set = set(train_symbols) train_set = set(train_symbols)
validation_set = set(validation_symbols) validation_set = set(validation_symbols)
test_set = set(test_symbols) test_set = set(test_symbols)
step = EIGHT_H_MS step = int(step_ms)
rows: list[pd.DataFrame] = [] rows: list[pd.DataFrame] = []
candidate_counts = {"train": 0, "validation": 0, "calibration": 0, "test": 0} candidate_counts = {"train": 0, "validation": 0, "calibration": 0, "test": 0}
selected_counts = {name: 0 for name in candidate_counts} selected_counts = {name: 0 for name in candidate_counts}
@@ -515,10 +517,13 @@ def _artifact(
manifest: dict[str, Any], manifest: dict[str, Any],
end_ms: int, end_ms: int,
test_start_ms: int, test_start_ms: int,
timeframe: str,
step_ms: int,
) -> dict[str, Any]: ) -> dict[str, Any]:
return { return {
"version": 4, "version": 4,
"timeframe": "8h", "timeframe": timeframe,
"step_ms": step_ms,
"window": WINDOW, "window": WINDOW,
"horizon": HORIZON, "horizon": HORIZON,
"target_kind": "signed_future_low_q20_return_atr_normalized", "target_kind": "signed_future_low_q20_return_atr_normalized",
@@ -572,10 +577,12 @@ def run_experiments(
) -> dict[str, Any]: ) -> dict[str, Any]:
started = time.monotonic() started = time.monotonic()
output_dir.mkdir(parents=True, exist_ok=True) 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) manifest = load_manifest(data_dir)
frames = load_raw_frames(data_dir, manifest) 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) test_start_ms = months_before(end_ms, test_months)
indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()} indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()}
raw_profiles = { raw_profiles = {
@@ -585,13 +592,14 @@ def run_experiments(
profiles = {symbol: profile for symbol, profile in raw_profiles.items() if profile is not None} 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) train_symbols, validation_symbols, test_symbols = split_holdout_symbols(profiles, seed)
scaler = fit_asset_scaler(profiles, train_symbols) 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( _progress(
f"dataset={len(frames)} symbols; train/validation/test symbols=" f"dataset={len(frames)} symbols; train/validation/test symbols="
f"{len(train_symbols)}/{len(validation_symbols)}/{len(test_symbols)}" f"{len(train_symbols)}/{len(validation_symbols)}/{len(test_symbols)}"
) )
stored_baseline = json.loads(baseline_report.read_text()) stored_baseline = json.loads(baseline_report.read_text())
stored_baseline_score = _baseline_score(stored_baseline) 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}") _progress(f"stored baseline validation selection score={stored_baseline_score:.8f}")
results: list[dict[str, Any]] = [] results: list[dict[str, Any]] = []
fitted_by_name: dict[str, tuple[FeatureVariant, dict[str, Any], pd.DataFrame, dict[str, dict[str, float]]]] = {} 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, validation_end_ms,
calibration_end_ms, calibration_end_ms,
volatility_rank_threshold, volatility_rank_threshold,
step_ms,
) )
filter_stats = dataset.attrs.get("volatility_filter", {}) filter_stats = dataset.attrs.get("volatility_filter", {})
_progress( _progress(
@@ -673,7 +682,7 @@ def run_experiments(
"validation_selection_score": comparison_baseline_score, "validation_selection_score": comparison_baseline_score,
"validation": next(item["validation"] for item in results if item["name"] == "baseline"), "validation": next(item["validation"] for item in results if item["name"] == "baseline"),
"test": baseline_test, "test": baseline_test,
"stored_test": stored_baseline["test"], "stored_test": stored_test,
} }
best_test = _evaluate_test(best_dataset, best_fitted) best_test = _evaluate_test(best_dataset, best_fitted)
best_dir = output_dir / "best" best_dir = output_dir / "best"
@@ -687,13 +696,22 @@ def run_experiments(
manifest, manifest,
end_ms, end_ms,
test_start_ms, test_start_ms,
timeframe,
step_ms,
), ),
best_dir / "model.joblib", best_dir / "model.joblib",
) )
report = { report = {
"protocol": { "protocol": {
"data_dir": str(data_dir.resolve()), "data_dir": str(data_dir.resolve()),
"timeframe": timeframe,
"step_ms": step_ms,
"end_ms": end_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, "test_start_ms": test_start_ms,
"validation_end_ms": validation_end_ms, "validation_end_ms": validation_end_ms,
"calibration_end_ms": calibration_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: def _write_report_markdown(path: Path, report: dict[str, Any]) -> None:
baseline = report["baseline"] baseline = report["baseline"]
selected = report["selected"] selected = report["selected"]
protocol = report["protocol"]
lines = [ lines = [
"# Feature experiment report", "# 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"- 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: `{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}`", f"- selected test MAE/RMSE: `{selected['test']['mae_point']:.6f}` / `{selected['test']['rmse_point']:.6f}`",

View File

@@ -13,7 +13,7 @@ from urllib.parse import urlparse
import joblib 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, timeframe_step_ms
REPO_ROOT = Path(__file__).resolve().parents[1] 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 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( def ensure_database(
banbot_config: Path, banbot_config: Path,
dsn: str | None, dsn: str | None,
@@ -86,6 +92,9 @@ 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) 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", {})) trained_symbols = set(artifact_data.get("asset_profiles", {}))
if trained_symbols: if trained_symbols:
skipped_symbols = [ skipped_symbols = [
@@ -117,7 +126,7 @@ def build_predictions(data_dir: Path, artifact: Path, output_dir: Path) -> dict[
"symbol": result["symbol"], "symbol": result["symbol"],
"latest_bar_start_utc": _utc_label(int(result["timestamp"])), "latest_bar_start_utc": _utc_label(int(result["timestamp"])),
"close": result["close"], "close": result["close"],
"quote_volume_usdt_8h": float(latest["quote"]), volume_column: float(latest["quote"]),
"atr_rank_200_pct": ( "atr_rank_200_pct": (
result["volatility_rank_200"] * 100 result["volatility_rank_200"] * 100
if result["volatility_rank_200"] is not None 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: if index == 1 or index == total or index % 25 == 0:
print(f"[latest-predict] predicted {index}/{total}", flush=True) 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): for rank, row in enumerate(rows, 1):
row["rank"] = rank row["rank"] = rank
output_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True)
output_csv = output_dir / "predictions.csv" output_csv = output_dir / "predictions.csv"
with output_csv.open("w", newline="") as handle: 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.writeheader()
writer.writerows(rows) writer.writerows(rows)
summary = { summary = {
"model": str(artifact.resolve()), "model": str(artifact.resolve()),
"timeframe": timeframe,
"symbols": len(rows), "symbols": len(rows),
"skipped_symbols": skipped_symbols, "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),
@@ -176,6 +186,7 @@ def build_parser() -> argparse.ArgumentParser:
parser.add_argument("--dsn", default=None) parser.add_argument("--dsn", default=None)
parser.add_argument("--db-type", choices=["auto", "questdb", "timescale"], default="auto") parser.add_argument("--db-type", choices=["auto", "questdb", "timescale"], default="auto")
parser.add_argument("--exchange", default="binance") parser.add_argument("--exchange", default="binance")
parser.add_argument("--timeframe", choices=["8h", "1d"], default="8h")
return parser return parser
@@ -192,11 +203,12 @@ def main() -> None:
dsn=args.dsn, dsn=args.dsn,
db_type=args.db_type, db_type=args.db_type,
exchange=args.exchange, 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, history_months=args.history_months,
holdout_months=args.holdout_months, holdout_months=args.holdout_months,
min_age_days=args.min_age_days, min_age_days=args.min_age_days,
chunk_size=args.chunk_size, chunk_size=args.chunk_size,
timeframe=args.timeframe,
) )
print( print(
f"[latest-predict] exported {len(manifest['symbols'])} symbols; " f"[latest-predict] exported {len(manifest['symbols'])} symbols; "

View File

@@ -25,7 +25,12 @@ from sklearn.metrics import mean_absolute_error, mean_squared_error
EIGHT_H_MS = 8 * 60 * 60 * 1000 EIGHT_H_MS = 8 * 60 * 60 * 1000
DAY_MS = 24 * 60 * 60 * 1000
HOUR_MS = 60 * 60 * 1000 HOUR_MS = 60 * 60 * 1000
TIMEFRAME_STEP_MS = {
"8h": EIGHT_H_MS,
"1d": DAY_MS,
}
WINDOW = 18 WINDOW = 18
HORIZON = 6 HORIZON = 6
VOLATILITY_RANK_WINDOW = 200 VOLATILITY_RANK_WINDOW = 200
@@ -62,9 +67,21 @@ class Symbol:
combined: bool = False 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: 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): if isinstance(value, int):
return value return value
text = str(value).strip() 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: 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: def months_before(value_ms: int, months: int) -> int:
@@ -237,6 +254,7 @@ def run_banbot_export(
history_start_ms: int, history_start_ms: int,
end_ms: int, end_ms: int,
banbot_export_cmd: str = "go run ./cmd/cbpred-export", banbot_export_cmd: str = "go run ./cmd/cbpred-export",
timeframe: str = "8h",
) -> None: ) -> None:
if not symbols: if not symbols:
raise ValueError("no eligible symbols to export") raise ValueError("no eligible symbols to export")
@@ -253,6 +271,8 @@ def run_banbot_export(
str(history_start_ms), str(history_start_ms),
"--end-ms", "--end-ms",
str(end_ms), str(end_ms),
"--timeframe",
timeframe,
"--out-dir", "--out-dir",
str(out_dir), str(out_dir),
] ]
@@ -260,7 +280,9 @@ def run_banbot_export(
subprocess.run(command, cwd=project_dir, check=True) 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 columns = CSV_COLUMNS
values = list(rows) values = list(rows)
if not values: 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["buy_volume"] = frame["buy_volume"].fillna(0.0)
frame["trade_num"] = frame["trade_num"].fillna(0).astype(np.int64) 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["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]] = [] output: list[dict[str, Any]] = []
for bucket, group in frame.groupby("bucket", sort=True): for bucket, group in frame.groupby("bucket", sort=True):
group = group.sort_values("timestamp") group = group.sort_values("timestamp")
expected = bucket + np.arange(8, dtype=np.int64) * HOUR_MS expected = bucket + np.arange(bars_per_bucket, dtype=np.int64) * HOUR_MS
if len(group) != 8 or not np.array_equal(group["timestamp"].to_numpy(), expected): if len(group) != bars_per_bucket or not np.array_equal(
group["timestamp"].to_numpy(), expected
):
continue continue
output.append( output.append(
{ {
@@ -298,6 +322,10 @@ def aggregate_8h(rows: Iterable[tuple[Any, ...]]) -> pd.DataFrame:
return pd.DataFrame(output, columns=columns) 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: def symbol_filename(symbol: Symbol) -> str:
slug = symbol.symbol.replace("/", "_").replace(":", "_") slug = symbol.symbol.replace("/", "_").replace(":", "_")
return f"{symbol.sid}_{slug}.csv" return f"{symbol.sid}_{slug}.csv"
@@ -318,11 +346,13 @@ def prepare_data(
skip_download: bool = False, skip_download: bool = False,
banbot_cmd: str = "go run .", banbot_cmd: str = "go run .",
banbot_export_cmd: str = "go run ./cmd/cbpred-export", banbot_export_cmd: str = "go run ./cmd/cbpred-export",
timeframe: str = "8h",
) -> dict[str, Any]: ) -> dict[str, Any]:
step_ms = timeframe_step_ms(timeframe)
out_dir.mkdir(parents=True, exist_ok=True) out_dir.mkdir(parents=True, exist_ok=True)
raw_dir = out_dir / "raw" raw_dir = out_dir / "raw"
raw_dir.mkdir(parents=True, exist_ok=True) 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) history_start_ms = months_before(end_ms, history_months)
training_snapshot_ms = months_before(end_ms, holdout_months) training_snapshot_ms = months_before(end_ms, holdout_months)
database_url = load_database_url(banbot_config, dsn) database_url = load_database_url(banbot_config, dsn)
@@ -356,12 +386,13 @@ def prepare_data(
history_start_ms, history_start_ms,
end_ms, end_ms,
banbot_export_cmd, banbot_export_cmd,
timeframe,
) )
manifest_symbols: list[dict[str, Any]] = [] manifest_symbols: list[dict[str, Any]] = []
for symbol in symbols: for symbol in symbols:
path = raw_dir / symbol_filename(symbol) path = raw_dir / symbol_filename(symbol)
if not path.exists(): if not path.exists():
excluded.append({**symbol.__dict__, "reason": "no_complete_8h_bars"}) excluded.append({**symbol.__dict__, "reason": f"no_complete_{timeframe}_bars"})
continue continue
frame = pd.read_csv(path) frame = pd.read_csv(path)
frame["timestamp"] = pd.to_numeric(frame["timestamp"], errors="coerce") frame["timestamp"] = pd.to_numeric(frame["timestamp"], errors="coerce")
@@ -370,18 +401,21 @@ def prepare_data(
frame = frame[frame["timestamp"] < symbol.delist_ms] frame = frame[frame["timestamp"] < symbol.delist_ms]
frame.to_csv(path, index=False, float_format="%.12g") frame.to_csv(path, index=False, float_format="%.12g")
if frame.empty: 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) path.unlink(missing_ok=True)
continue continue
manifest_symbols.append({**symbol.__dict__, "path": str(path.relative_to(out_dir)), "bars": len(frame)}) manifest_symbols.append({**symbol.__dict__, "path": str(path.relative_to(out_dir)), "bars": len(frame)})
manifest = { manifest = {
"version": 2, "version": 2,
"timeframe": "8h", "timeframe": timeframe,
"storage_timeframe": "banbot_orm", "storage_timeframe": "banbot_orm",
"step_ms": EIGHT_H_MS, "step_ms": step_ms,
"end_ms": end_ms, "end_ms": end_ms,
"history_start_ms": history_start_ms, "history_start_ms": history_start_ms,
"training_snapshot_ms": training_snapshot_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, "db_type": actual_db_type,
"exchange": exchange, "exchange": exchange,
"symbols": manifest_symbols, "symbols": manifest_symbols,
@@ -727,6 +761,7 @@ def build_feature_dataset(
validation_end_ms: int | None = None, validation_end_ms: int | None = None,
calibration_end_ms: int | None = None, calibration_end_ms: int | None = None,
volatility_rank_threshold: float = VOLATILITY_RANK_THRESHOLD, volatility_rank_threshold: float = VOLATILITY_RANK_THRESHOLD,
step_ms: int = EIGHT_H_MS,
) -> tuple[pd.DataFrame, list[str], dict[str, dict[str, float]]]: ) -> tuple[pd.DataFrame, list[str], dict[str, dict[str, float]]]:
if not 0.0 <= volatility_rank_threshold <= 1.0: if not 0.0 <= volatility_rank_threshold <= 1.0:
raise ValueError("volatility_rank_threshold must be between 0 and 1") raise ValueError("volatility_rank_threshold must be between 0 and 1")
@@ -745,7 +780,7 @@ def build_feature_dataset(
rows: list[pd.DataFrame] = [] rows: list[pd.DataFrame] = []
candidate_counts = {"train": 0, "validation": 0, "calibration": 0, "test": 0, "recent": 0} candidate_counts = {"train": 0, "validation": 0, "calibration": 0, "test": 0, "recent": 0}
selected_counts = {name: 0 for name in candidate_counts} 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 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 []) train_set = set(train_symbols or [])
validation_set = set(validation_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 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.""" """Use the first 30% for model selection, next 20% for calibration, last 50% for final test."""
duration = end_ms - start_ms duration = end_ms - start_ms
validation_end = floor_to_8h(start_ms + int(duration * 0.30)) validation_end = start_ms + (int(duration * 0.30) // step_ms) * step_ms
calibration_end = floor_to_8h(start_ms + int(duration * 0.50)) calibration_end = start_ms + (int(duration * 0.50) // step_ms) * step_ms
validation_end = max(start_ms + EIGHT_H_MS, min(validation_end, end_ms - 2 * EIGHT_H_MS)) validation_end = max(start_ms + step_ms, min(validation_end, end_ms - 2 * step_ms))
calibration_end = max(validation_end + EIGHT_H_MS, min(calibration_end, end_ms - EIGHT_H_MS)) calibration_end = max(validation_end + step_ms, min(calibration_end, end_ms - step_ms))
return validation_end, calibration_end return validation_end, calibration_end
@@ -1182,12 +1219,14 @@ def run_training(
) -> dict[str, Any]: ) -> dict[str, Any]:
if not 0.0 <= volatility_rank_threshold <= 1.0: if not 0.0 <= volatility_rank_threshold <= 1.0:
raise ValueError("volatility_rank_threshold must be between 0 and 1") 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) manifest = load_manifest(data_dir)
frames = load_raw_frames(data_dir, manifest) frames = load_raw_frames(data_dir, manifest)
if len(frames) < 3: if len(frames) < 3:
raise ValueError("at least 3 symbols with raw data are required") 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) test_start_ms = months_before(end_ms, test_months)
profile_indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()} profile_indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()}
raw_profiles = { raw_profiles = {
@@ -1199,7 +1238,7 @@ def run_training(
raise ValueError("at least 3 symbols with enough pre-test history are required") raise ValueError("at least 3 symbols with enough pre-test history are required")
train_symbols, validation_symbols, test_symbols = split_holdout_symbols(profiles, seed) train_symbols, validation_symbols, test_symbols = split_holdout_symbols(profiles, seed)
asset_scaler = fit_asset_scaler(profiles, train_symbols) 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( _progress(
f"training: {len(frames)} symbols, {len(profiles)} profiles; " f"training: {len(frames)} symbols, {len(profiles)} profiles; "
f"train/validation/test symbols={len(train_symbols)}/{len(validation_symbols)}/{len(test_symbols)}" f"train/validation/test symbols={len(train_symbols)}/{len(validation_symbols)}/{len(test_symbols)}"
@@ -1226,6 +1265,7 @@ def run_training(
validation_end_ms, validation_end_ms,
calibration_end_ms, calibration_end_ms,
volatility_rank_threshold, volatility_rank_threshold,
step_ms,
) )
filter_stats = dataset.attrs.get("volatility_filter", {}) filter_stats = dataset.attrs.get("volatility_filter", {})
_progress( _progress(
@@ -1367,7 +1407,8 @@ def run_training(
artifact_dir.mkdir(parents=True, exist_ok=True) artifact_dir.mkdir(parents=True, exist_ok=True)
artifact = { artifact = {
"version": 3, "version": 3,
"timeframe": "8h", "timeframe": timeframe,
"step_ms": step_ms,
"window": WINDOW, "window": WINDOW,
"horizon": HORIZON, "horizon": HORIZON,
"target_kind": TARGET_KIND, "target_kind": TARGET_KIND,
@@ -1396,6 +1437,15 @@ def run_training(
joblib.dump(artifact, artifact_dir / "model.joblib") joblib.dump(artifact, artifact_dir / "model.joblib")
report = { report = {
"protocol": { "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, "train_anchor_before": test_start_ms,
"recent_window_start": test_start_ms, "recent_window_start": test_start_ms,
"recent_window_end": end_ms, "recent_window_end": end_ms,
@@ -1485,6 +1535,8 @@ def predict_from_csv(
raise ValueError( raise ValueError(
"artifact uses an unsupported target or requires full-universe feature context" "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) frame = pd.read_csv(csv_path)
for column in CSV_COLUMNS: for column in CSV_COLUMNS:
frame[column] = pd.to_numeric(frame[column], errors="coerce") 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") raise ValueError("--at is earlier than the model training snapshot")
matches = np.flatnonzero(frame["timestamp"].to_numpy(dtype=np.int64) == int(at_ms)) matches = np.flatnonzero(frame["timestamp"].to_numpy(dtype=np.int64) == int(at_ms))
if len(matches) == 0: 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]) anchor_index = int(matches[-1])
if symbol is None: if symbol is None:
for candidate, path in artifact.get("symbol_files", {}).items(): for candidate, path in artifact.get("symbol_files", {}).items():
@@ -1519,6 +1571,8 @@ def predict_from_csv(
) )
response = { response = {
"timestamp": int(frame.iloc[anchor_index]["timestamp"]), "timestamp": int(frame.iloc[anchor_index]["timestamp"]),
"timeframe": timeframe,
"step_ms": step_ms,
"close": float(frame.iloc[anchor_index]["close"]), "close": float(frame.iloc[anchor_index]["close"]),
"model": artifact["model_name"], "model": artifact["model_name"],
"feature_stage": feature_stage, "feature_stage": feature_stage,

View File

@@ -25,21 +25,22 @@ func main() {
pairs := flag.String("pairs", "", "comma-separated symbols") pairs := flag.String("pairs", "", "comma-separated symbols")
startMS := flag.Int64("start-ms", 0, "inclusive start timestamp in milliseconds") startMS := flag.Int64("start-ms", 0, "inclusive start timestamp in milliseconds")
endMS := flag.Int64("end-ms", 0, "exclusive end 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") outDir := flag.String("out-dir", "", "CSV output directory")
flag.Parse() 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() flag.Usage()
os.Exit(2) os.Exit(2)
} }
args := &config.CmdArgs{ args := &config.CmdArgs{
Configs: config.ArrString{*configPath}, Configs: config.ArrString{*configPath},
DataDir: *dataDir, DataDir: *dataDir,
NoDefault: true, NoDefault: true,
RawPairs: *pairs, RawPairs: *pairs,
TimeStart: strconv.FormatInt(*startMS, 10), TimeStart: strconv.FormatInt(*startMS, 10),
TimeEnd: strconv.FormatInt(*endMS, 10), TimeEnd: strconv.FormatInt(*endMS, 10),
LogLevel: "error", LogLevel: "error",
} }
if err := biz.SetupComs(args); err != nil { if err := biz.SetupComs(args); err != nil {
fail(err) fail(err)
@@ -58,7 +59,7 @@ func main() {
if err != nil { if err != nil {
fail(err) 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 { if err != nil {
fail(err) fail(err)
} }

77
docs/1d_2026-08-24.md Normal file
View File

@@ -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 的可见性告警,但本次导出命令成功完成,最终文件连续性检查全部通过。

View File

@@ -123,3 +123,11 @@ price_level = close × (1 + future_low_return)
- 模型只对保存了训练画像且有足够历史的品种预测。当前 ATR 排名至少需要 200 根 8H K 线。 - 模型只对保存了训练画像且有足够历史的品种预测。当前 ATR 排名至少需要 200 根 8H K 线。
- `atr_rank_200_pct` 是状态描述,不代表模型只预测高波动时刻;正式阈值为 `0.0`。 - `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 根预测跨度不同,不应直接当作同一持有期的收益比较。

View File

@@ -31,7 +31,7 @@ python3 -m venv .venv
1. 检查配置中的 QuestDB 是否可连接;QuestDB 默认不可用时,通过 banbot 的导出命令触发其启动流程。 1. 检查配置中的 QuestDB 是否可连接;QuestDB 默认不可用时,通过 banbot 的导出命令触发其启动流程。
2. 从数据库筛选 Binance `linear` 市场、结算币为 USDT、上市至少 730 天且未退市的品种。 2. 从数据库筛选 Binance `linear` 市场、结算币为 USDT、上市至少 730 天且未退市的品种。
3. 调用 banbot 下载最近 36 个月的 1H K 线。 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`。 5. 写入 `data/manifest.json` 和 `data/raw/*.csv`。
6. 加载 `artifacts/v1/model.joblib`,对清单中的每个品种使用最新完整 8H K 线预测。 6. 加载 `artifacts/v1/model.joblib`,对清单中的每个品种使用最新完整 8H K 线预测。
7. 按最新 8H USDT 成交额降序写入 `predictions/latest_v1/predictions.csv`,并写入 `report.json`。 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 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` 记录筛选出的品种、数据范围、数据库类型和被排除的原因。 `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 ```bash
@@ -96,9 +105,9 @@ python3 -m venv .venv
--symbol BTC/USDT:USDT --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 启动。 - QuestDB 连接失败:检查 banbot 配置中的数据库地址、QuestDB 服务和端口;一键脚本只对默认 QuestDB 端口尝试触发 banbot 启动。
- `missing data/manifest.json`:先运行 `prepare`,或确认一键脚本没有在下载/导出阶段失败。 - `missing data/manifest.json`:先运行 `prepare`,或确认一键脚本没有在下载/导出阶段失败。
- 某品种预测为空:检查 `eligible_for_model` 和 `skip_reason`,通常是 200 根 ATR 历史不足或该品种没有正式模型画像。 - 某品种预测为空:检查 `eligible_for_model` 和 `skip_reason`,通常是 200 根 ATR 历史不足或该品种没有正式模型画像。
- 数据只有部分 8H K 线:导出会丢弃不连续或不完整的 8 根 1H 桶,这是为了避免特征和未来标签跨越缺口。 - 数据只有部分 K 线:导出和训练会丢弃不连续或不完整的 bar,这是为了避免特征和未来标签跨越缺口。
## 验证 ## 验证

View File

@@ -15,6 +15,8 @@ class ArtifactPathTests(unittest.TestCase):
self.assertEqual(train.artifact_dir, Path("artifacts/v1")) self.assertEqual(train.artifact_dir, Path("artifacts/v1"))
self.assertEqual(run.artifact_dir, Path("artifacts/v1")) self.assertEqual(run.artifact_dir, Path("artifacts/v1"))
self.assertEqual(predict.artifact, Path("artifacts/v1/model.joblib")) self.assertEqual(predict.artifact, Path("artifacts/v1/model.joblib"))
daily = parser.parse_args(["prepare", "--timeframe", "1d"])
self.assertEqual(daily.timeframe, "1d")
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -6,7 +6,7 @@ from unittest.mock import patch
import pandas as pd 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): 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): def test_batch_skips_symbols_without_training_profiles(self):
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
root = Path(directory) root = Path(directory)

View File

@@ -10,7 +10,9 @@ import pandas as pd
from cbpred.pipeline import ( from cbpred.pipeline import (
CSV_COLUMNS, CSV_COLUMNS,
DAY_MS,
EIGHT_H_MS, EIGHT_H_MS,
aggregate_ohlcv,
aggregate_8h, aggregate_8h,
build_feature_dataset, build_feature_dataset,
compute_indicators, compute_indicators,
@@ -45,6 +47,33 @@ class PipelineTests(unittest.TestCase):
self.assertEqual(frame.iloc[0]["timestamp"], 0) self.assertEqual(frame.iloc[0]["timestamp"], 0)
self.assertAlmostEqual(frame.iloc[0]["quote"], sum(row[6] for row in rows[:8])) 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): def test_future_target_changes_but_past_features_do_not(self):
start = 0 start = 0
raw = aggregate_8h(make_hour_rows(start, 8 * 70)) raw = aggregate_8h(make_hour_rows(start, 8 * 70))