From 79b02d1c7d56bd638c33114d307db4071fde8258 Mon Sep 17 00:00:00 2001 From: anyongjin Date: Thu, 17 Sep 2026 22:02:51 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cbpred/latest_predict.py | 102 +++++++++++++++++++++++++-------- cbpred/pipeline.py | 70 ++++++++++++++-------- cmd/cbpred-export/main.go | 9 +++ cmd/cbpred-export/main_test.go | 16 ++++++ go.mod | 8 ++- go.sum | 8 +-- tests/test_latest_predict.py | 28 +++++++++ tests/test_pipeline.py | 27 +++++++++ 8 files changed, 213 insertions(+), 55 deletions(-) create mode 100644 cmd/cbpred-export/main_test.go diff --git a/cbpred/latest_predict.py b/cbpred/latest_predict.py index 91df834..94679db 100644 --- a/cbpred/latest_predict.py +++ b/cbpred/latest_predict.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse import csv +import fcntl import json import socket import subprocess @@ -53,6 +54,47 @@ def default_output_dir(timeframe: str) -> Path: return REPO_ROOT / f"predictions/latest_v1_{timeframe}" +def default_data_dir(timeframe: str) -> Path: + timeframe_step_ms(timeframe) + return REPO_ROOT / ("data" if timeframe == "8h" else "data_1d") + + +def default_artifact(timeframe: str) -> Path: + timeframe_step_ms(timeframe) + return REPO_ROOT / ( + "artifacts/v1/model.joblib" if timeframe == "8h" else "artifacts_1d/best/model.joblib" + ) + + +def _validate_artifact_timeframe(artifact: Path, timeframe: str) -> None: + artifact_timeframe = str(joblib.load(artifact).get("timeframe", "8h")) + if artifact_timeframe != timeframe: + raise ValueError( + f"artifact {artifact} is for {artifact_timeframe}, not requested timeframe {timeframe}" + ) + + +def _validate_freshness( + manifest: dict[str, object], prediction_items: list[dict[str, object]], data_dir: Path, timeframe: str +) -> None: + end_ms = manifest.get("end_ms") + if end_ms is None: + return + expected = int(end_ms) - timeframe_step_ms(timeframe) + stale: list[str] = [] + for item in prediction_items: + frame = pd.read_csv(data_dir / str(item["path"]), usecols=["timestamp"]) + latest = int(pd.to_numeric(frame["timestamp"], errors="coerce").max()) + if latest != expected: + stale.append(f"{item['symbol']}={_utc_label(latest)}") + if stale: + sample = ", ".join(stale[:5]) + raise RuntimeError( + f"incomplete {timeframe} data: expected latest bar {_utc_label(expected)}; " + f"{len(stale)}/{len(prediction_items)} prediction symbols differ ({sample})" + ) + + def ensure_database( banbot_config: Path, dsn: str | None, @@ -98,6 +140,11 @@ def build_predictions(data_dir: Path, artifact: Path, output_dir: Path) -> dict[ manifest = json.loads((data_dir / "manifest.json").read_text()) artifact_data = joblib.load(artifact) timeframe = str(manifest.get("timeframe", artifact_data.get("timeframe", "8h"))) + artifact_timeframe = str(artifact_data.get("timeframe", "8h")) + if artifact_timeframe != timeframe: + raise ValueError( + f"artifact {artifact} is for {artifact_timeframe}, not exported timeframe {timeframe}" + ) output_columns = output_columns_for_timeframe(timeframe) volume_column = output_columns[4] trained_symbols = set(artifact_data.get("asset_profiles", {})) @@ -113,6 +160,7 @@ def build_predictions(data_dir: Path, artifact: Path, output_dir: Path) -> dict[ else: skipped_symbols = [] prediction_items = manifest["symbols"] + _validate_freshness(manifest, prediction_items, data_dir, timeframe) if skipped_symbols: print( f"[latest-predict] skipped {len(skipped_symbols)} symbols without stored training profiles", @@ -180,8 +228,8 @@ def build_parser() -> argparse.ArgumentParser: type=Path, default=Path("/data/ban/banbot/biz/config.yml"), ) - parser.add_argument("--data-dir", type=Path, default=REPO_ROOT / "data") - parser.add_argument("--artifact", type=Path, default=REPO_ROOT / "artifacts/v1/model.joblib") + parser.add_argument("--data-dir", type=Path, default=None) + parser.add_argument("--artifact", type=Path, default=None) parser.add_argument( "--output-dir", type=Path, @@ -202,32 +250,36 @@ def build_parser() -> argparse.ArgumentParser: def main() -> None: args = build_parser().parse_args() - data_dir = args.data_dir.resolve() - artifact = args.artifact.resolve() + data_dir = (args.data_dir or default_data_dir(args.timeframe)).resolve() + artifact = (args.artifact or default_artifact(args.timeframe)).resolve() output_dir = ( args.output_dir if args.output_dir is not None else default_output_dir(args.timeframe) ).resolve() - ensure_database(args.banbot_config, args.dsn) - manifest = prepare_data( - out_dir=data_dir, - banbot_dir=args.banbot_dir, - banbot_config=args.banbot_config, - dsn=args.dsn, - db_type=args.db_type, - exchange=args.exchange, - 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; " - f"excluded {len(manifest['excluded'])}", - flush=True, - ) - summary = build_predictions(data_dir, artifact, output_dir) + _validate_artifact_timeframe(artifact, args.timeframe) + lock_path = args.banbot_config.parent / ".cbpred-latest-predict.lock" + with lock_path.open("a+") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + ensure_database(args.banbot_config, args.dsn) + manifest = prepare_data( + out_dir=data_dir, + banbot_dir=args.banbot_dir, + banbot_config=args.banbot_config, + dsn=args.dsn, + db_type=args.db_type, + exchange=args.exchange, + 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; " + f"excluded {len(manifest['excluded'])}", + flush=True, + ) + summary = build_predictions(data_dir, artifact, output_dir) print(json.dumps(summary, indent=2, ensure_ascii=False)) diff --git a/cbpred/pipeline.py b/cbpred/pipeline.py index c6b629a..82b84f2 100644 --- a/cbpred/pipeline.py +++ b/cbpred/pipeline.py @@ -4,6 +4,7 @@ import calendar import json import math import os +import re import shlex import subprocess import sys @@ -50,6 +51,8 @@ CSV_COLUMNS = [ "buy_volume", "trade_num", ] +INVALID_SYMBOL_RE = re.compile(r"\[InvalidSymbol\]\s+symbol\s+(\S+)\s+not found") +LINEAR_DOWNLOAD_CONFIG = "market_type: linear\npairlists: []\npairmgr:\n force_filters: false\n" def _progress(message: str) -> None: @@ -217,33 +220,53 @@ def run_banbot_download( end_ms: int, chunk_size: int = 100, banbot_cmd: str = "go run .", -) -> None: +) -> tuple[list[Symbol], list[Symbol]]: if not symbols: raise ValueError("no eligible symbols to download") base = shlex.split(banbot_cmd) + downloaded: list[Symbol] = [] + unavailable: list[Symbol] = [] for offset in range(0, len(symbols), chunk_size): chunk = symbols[offset : offset + chunk_size] - pairs = ",".join(item.symbol for item in chunk) - command = base + [ - "kline", - "down", - "--config", - str(config_path), - "--datadir", - str(config_path.parent), - "--no-default", - "--config-data", - "pairlists: []\npairmgr:\n force_filters: false\n", - "--pairs", - pairs, - "--timeframes", - "1h", - "--timestart", - str(history_start_ms), - "--timeend", - str(end_ms), - ] - subprocess.run(command, cwd=banbot_dir, check=True) + while chunk: + pairs = ",".join(item.symbol for item in chunk) + command = base + [ + "kline", + "down", + "--config", + str(config_path), + "--datadir", + str(config_path.parent), + "--no-default", + "--config-data", + LINEAR_DOWNLOAD_CONFIG, + "--pairs", + pairs, + "--timeframes", + "1h", + "--timestart", + str(history_start_ms), + "--timeend", + str(end_ms), + ] + try: + subprocess.run(command, cwd=banbot_dir, check=True, stderr=subprocess.PIPE, text=True) + except subprocess.CalledProcessError as exc: + invalid_symbols = set(INVALID_SYMBOL_RE.findall(exc.stderr or "")) + rejected = [item for item in chunk if item.symbol in invalid_symbols] + if not rejected: + if exc.stderr: + sys.stderr.write(exc.stderr) + raise + _progress( + "skipping unavailable symbols: " + ", ".join(item.symbol for item in rejected) + ) + unavailable.extend(rejected) + chunk = [item for item in chunk if item not in rejected] + else: + downloaded.extend(chunk) + break + return downloaded, unavailable def run_banbot_export( @@ -369,7 +392,7 @@ def prepare_data( finally: conn.close() if not skip_download: - run_banbot_download( + symbols, unavailable = run_banbot_download( symbols, banbot_dir, banbot_config, @@ -378,6 +401,7 @@ def prepare_data( chunk_size, banbot_cmd, ) + excluded.extend({**symbol.__dict__, "reason": "unavailable_on_exchange"} for symbol in unavailable) run_banbot_export( symbols, banbot_dir, diff --git a/cmd/cbpred-export/main.go b/cmd/cbpred-export/main.go index 63de2c0..f5ce3d1 100644 --- a/cmd/cbpred-export/main.go +++ b/cmd/cbpred-export/main.go @@ -13,6 +13,7 @@ import ( "github.com/banbox/banbot/config" "github.com/banbox/banbot/core" "github.com/banbox/banbot/orm" + "github.com/banbox/banexg/errs" ) var columns = []string{ @@ -27,6 +28,7 @@ func main() { 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") + skipUnavailable := flag.Bool("skip-unavailable", true, "skip pairs unavailable on the exchange and absent from the local cache") flag.Parse() if (*timeframe != "8h" && *timeframe != "1d") || *configPath == "" || *dataDir == "" || *pairs == "" || *outDir == "" || *startMS <= 0 || *endMS <= *startMS { flag.Usage() @@ -57,6 +59,9 @@ func main() { } exs, err := orm.GetExSymbolCur(pair) if err != nil { + if *skipUnavailable && isUnavailablePair(err) { + continue + } fail(err) } _, rows, err := orm.GetSeries(exs, *timeframe, *startMS, *endMS, 0, false) @@ -70,6 +75,10 @@ func main() { } } +func isUnavailablePair(err *errs.Error) bool { + return err != nil && err.Code == errs.CodeNoMarketForPair +} + func writeCSV(path string, rows []*orm.DataSeries, exs *orm.ExSymbol) error { file, err := os.Create(path) if err != nil { diff --git a/cmd/cbpred-export/main_test.go b/cmd/cbpred-export/main_test.go new file mode 100644 index 0000000..6b77ff0 --- /dev/null +++ b/cmd/cbpred-export/main_test.go @@ -0,0 +1,16 @@ +package main + +import ( + "testing" + + "github.com/banbox/banexg/errs" +) + +func TestIsUnavailablePair(t *testing.T) { + if !isUnavailablePair(errs.NewMsg(errs.CodeNoMarketForPair, "missing")) { + t.Fatal("NoMarketForPair should be skipped") + } + if isUnavailablePair(errs.NewMsg(errs.CodeInvalidData, "bad data")) { + t.Fatal("non-market errors must not be skipped") + } +} diff --git a/go.mod b/go.mod index 1eab93b..a6c94e3 100644 --- a/go.mod +++ b/go.mod @@ -2,12 +2,14 @@ module cbpred-export go 1.24.0 -require github.com/banbox/banbot v0.0.0 +require ( + github.com/banbox/banbot v0.0.0 + github.com/banbox/banexg v0.2.64 +) require ( github.com/bahlo/generic-list-go v0.2.0 // indirect - github.com/banbox/banexg v0.2.63 // indirect - github.com/banbox/banta v0.3.7 // indirect + github.com/banbox/banta v0.3.8 // indirect github.com/banbox/bntp v0.1.0 // indirect github.com/banbox/cron/v3 v3.1.2 // indirect github.com/beevik/ntp v1.4.3 // indirect diff --git a/go.sum b/go.sum index 85e95c2..20cf580 100644 --- a/go.sum +++ b/go.sum @@ -1,10 +1,10 @@ github.com/airbrake/gobrake v3.6.1+incompatible/go.mod h1:wM4gu3Cn0W0K7GUuVWnlXZU11AGBXMILnrdOU8Kn00o= github.com/bahlo/generic-list-go v0.2.0 h1:5sz/EEAK+ls5wF+NeqDpk5+iNdMDXrh3z3nPnH1Wvgk= github.com/bahlo/generic-list-go v0.2.0/go.mod h1:2KvAjgMlE5NNynlg/5iLrrCCZ2+5xWbdbCW3pNTGyYg= -github.com/banbox/banexg v0.2.63 h1:G93/8s+/uddzie+ckMQ0T+AQq+9ULoCs1tp9CsBl4KM= -github.com/banbox/banexg v0.2.63/go.mod h1:T9JWRGblxAibQjwKK6VABG+r87y1LRnyYe7sz5aBnR4= -github.com/banbox/banta v0.3.7 h1:R1xVVtds0Temlc1fATX08uG7rR0pbukr3L6FikyKscE= -github.com/banbox/banta v0.3.7/go.mod h1:43XoXngLPip/pX1OvohnNUScScqb1/jtKxVqgltwZUU= +github.com/banbox/banexg v0.2.64 h1:jnHCzWYCnF9MVEa6RWD4CJfMpOS04Yq5RXbA1Dh6mxY= +github.com/banbox/banexg v0.2.64/go.mod h1:T9JWRGblxAibQjwKK6VABG+r87y1LRnyYe7sz5aBnR4= +github.com/banbox/banta v0.3.8 h1:P/sfgCd5myy/exWU3AKlxLV45mhMdXTaRSusCSiSjKs= +github.com/banbox/banta v0.3.8/go.mod h1:43XoXngLPip/pX1OvohnNUScScqb1/jtKxVqgltwZUU= github.com/banbox/bntp v0.1.0 h1:kZROpbb3OME8XM8QeM+2SeQWQrHRDFN05okr8OITKAU= github.com/banbox/bntp v0.1.0/go.mod h1:w0WP7GROKQo07wHFBm/FVgH4xlZuOYCoro0ky4iTC3M= github.com/banbox/cron/v3 v3.1.2 h1:gBNnz/OBxIMU7iBCtObZKhb85rzbaXr8AZLZb40CgeM= diff --git a/tests/test_latest_predict.py b/tests/test_latest_predict.py index 598e76a..be6e016 100644 --- a/tests/test_latest_predict.py +++ b/tests/test_latest_predict.py @@ -10,6 +10,8 @@ from cbpred.latest_predict import ( OUTPUT_COLUMNS, REPO_ROOT, build_predictions, + default_artifact, + default_data_dir, default_output_dir, output_columns_for_timeframe, ) @@ -40,6 +42,32 @@ class LatestPredictTests(unittest.TestCase): self.assertEqual(default_output_dir("8h"), REPO_ROOT / "predictions/latest_v1_8h") self.assertEqual(default_output_dir("1d"), REPO_ROOT / "predictions/latest_v1_1d") + def test_default_inputs_follow_timeframe(self): + self.assertEqual(default_data_dir("8h"), REPO_ROOT / "data") + self.assertEqual(default_data_dir("1d"), REPO_ROOT / "data_1d") + self.assertEqual(default_artifact("8h"), REPO_ROOT / "artifacts/v1/model.joblib") + self.assertEqual(default_artifact("1d"), REPO_ROOT / "artifacts_1d/best/model.joblib") + + def test_batch_rejects_stale_prediction_data(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "item.csv").write_text("timestamp,quote,close\n0,1,1\n") + (root / "manifest.json").write_text( + json.dumps( + { + "timeframe": "1d", + "end_ms": 2 * 24 * 60 * 60 * 1000, + "symbols": [{"symbol": "TRAIN/USDT:USDT", "path": "item.csv"}], + } + ) + ) + with patch( + "cbpred.latest_predict.joblib.load", + return_value={"timeframe": "1d", "asset_profiles": {"TRAIN/USDT:USDT": {}}}, + ): + with self.assertRaisesRegex(RuntimeError, "incomplete 1d data"): + build_predictions(root, root / "model.joblib", root / "output") + 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 b647256..5fe4bb5 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -1,9 +1,11 @@ from __future__ import annotations import json +import subprocess import tempfile import unittest from pathlib import Path +from unittest.mock import patch import numpy as np import pandas as pd @@ -21,6 +23,8 @@ from cbpred.pipeline import ( _window_feature_names, _window_features, run_training, + Symbol, + run_banbot_download, ) @@ -40,6 +44,29 @@ def make_hour_rows(start_ms: int, hours: int, phase: float = 0.0): class PipelineTests(unittest.TestCase): + def test_download_skips_invalid_current_symbol_and_retries_chunk(self): + valid = Symbol(1, "binance", "linear", "BTC/USDT:USDT", 1, 0) + stale = Symbol(2, "binance", "linear", "SXP/USDT:USDT", 1, 0) + calls = [] + + def run(command, **kwargs): + calls.append(command) + if len(calls) == 1: + raise subprocess.CalledProcessError( + 1, command, stderr="Error: [InvalidSymbol] symbol SXP/USDT:USDT not found\n" + ) + + with patch("cbpred.pipeline.subprocess.run", side_effect=run): + downloaded, unavailable = run_banbot_download( + [valid, stale], Path("/banbot"), Path("/banbot/config.yml"), 1, 2 + ) + + self.assertEqual(downloaded, [valid]) + self.assertEqual(unavailable, [stale]) + self.assertEqual(calls[0][calls[0].index("--pairs") + 1], "BTC/USDT:USDT,SXP/USDT:USDT") + self.assertEqual(calls[1][calls[1].index("--pairs") + 1], "BTC/USDT:USDT") + self.assertIn("market_type: linear", calls[0][calls[0].index("--config-data") + 1]) + def test_aggregate_8h_drops_incomplete_buckets(self): rows = make_hour_rows(0, 17) frame = aggregate_8h(rows)