更新脚本

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("--data-dir", type=Path, default=None)
-    parser.add_argument("--artifact", type=Path, default=REPO_ROOT / "artifacts/v1/model.joblib")
+    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()
+    data_dir = (args.data_dir or default_data_dir(args.timeframe)).resolve()
-    artifact = args.artifact.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)
+    _validate_artifact_timeframe(artifact, args.timeframe)
-    manifest = prepare_data(
+    lock_path = args.banbot_config.parent / ".cbpred-latest-predict.lock"
-        out_dir=data_dir,
+    with lock_path.open("a+") as lock:
-        banbot_dir=args.banbot_dir,
+        fcntl.flock(lock, fcntl.LOCK_EX)
-        banbot_config=args.banbot_config,
+        ensure_database(args.banbot_config, args.dsn)
-        dsn=args.dsn,
+        manifest = prepare_data(
-        db_type=args.db_type,
+            out_dir=data_dir,
-        exchange=args.exchange,
+            banbot_dir=args.banbot_dir,
-        end_ms=parse_time_ms(args.end, args.timeframe) if args.end else None,
+            banbot_config=args.banbot_config,
-        history_months=args.history_months,
+            dsn=args.dsn,
-        holdout_months=args.holdout_months,
+            db_type=args.db_type,
-        min_age_days=args.min_age_days,
+            exchange=args.exchange,
-        chunk_size=args.chunk_size,
+            end_ms=parse_time_ms(args.end, args.timeframe) if args.end else None,
-        timeframe=args.timeframe,
+            history_months=args.history_months,
-    )
+            holdout_months=args.holdout_months,
-    print(
+            min_age_days=args.min_age_days,
-        f"[latest-predict] exported {len(manifest['symbols'])} symbols; "
+            chunk_size=args.chunk_size,
-        f"excluded {len(manifest['excluded'])}",
+            timeframe=args.timeframe,
-        flush=True,
+        )
-    )
+        print(
-    summary = build_predictions(data_dir, artifact, output_dir)
+            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))
 
 

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)
+        while chunk:
-        command = base + [
+            pairs = ",".join(item.symbol for item in chunk)
-            "kline",
+            command = base + [
-            "down",
+                "kline",
-            "--config",
+                "down",
-            str(config_path),
+                "--config",
-            "--datadir",
+                str(config_path),
-            str(config_path.parent),
+                "--datadir",
-            "--no-default",
+                str(config_path.parent),
-            "--config-data",
+                "--no-default",
-            "pairlists: []\npairmgr:\n  force_filters: false\n",
+                "--config-data",
-            "--pairs",
+                LINEAR_DOWNLOAD_CONFIG,
-            pairs,
+                "--pairs",
-            "--timeframes",
+                pairs,
-            "1h",
+                "--timeframes",
-            "--timestart",
+                "1h",
-            str(history_start_ms),
+                "--timestart",
-            "--timeend",
+                str(history_start_ms),
-            str(end_ms),
+                "--timeend",
-        ]
+                str(end_ms),
-        subprocess.run(command, cwd=banbot_dir, check=True)
+            ]
+            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,

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 {

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")
+	}
+}

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.8 // indirect
-	github.com/banbox/banta v0.3.7 // 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

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.64 h1:jnHCzWYCnF9MVEa6RWD4CJfMpOS04Yq5RXbA1Dh6mxY=
-github.com/banbox/banexg v0.2.63/go.mod h1:T9JWRGblxAibQjwKK6VABG+r87y1LRnyYe7sz5aBnR4=
+github.com/banbox/banexg v0.2.64/go.mod h1:T9JWRGblxAibQjwKK6VABG+r87y1LRnyYe7sz5aBnR4=
-github.com/banbox/banta v0.3.7 h1:R1xVVtds0Temlc1fATX08uG7rR0pbukr3L6FikyKscE=
+github.com/banbox/banta v0.3.8 h1:P/sfgCd5myy/exWU3AKlxLV45mhMdXTaRSusCSiSjKs=
-github.com/banbox/banta v0.3.7/go.mod h1:43XoXngLPip/pX1OvohnNUScScqb1/jtKxVqgltwZUU=
+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=

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)

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)