支持一键预测脚本
This commit is contained in:
190
cbpred/latest_predict.py
Normal file
190
cbpred/latest_predict.py
Normal file
@@ -0,0 +1,190 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import socket
|
||||
import subprocess
|
||||
import tempfile
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from .pipeline import load_database_url, parse_time_ms, predict_from_csv, prepare_data
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
OUTPUT_COLUMNS = (
|
||||
"rank",
|
||||
"symbol",
|
||||
"latest_bar_start_utc",
|
||||
"close",
|
||||
"quote_volume_usdt_8h",
|
||||
"atr_rank_200_pct",
|
||||
"drawdown_point_pct",
|
||||
"drawdown_cap_80pct_pct",
|
||||
"price_level_reached_80pct",
|
||||
"price_level_cap_80pct",
|
||||
)
|
||||
|
||||
|
||||
def _utc_label(timestamp_ms: int) -> str:
|
||||
return datetime.fromtimestamp(timestamp_ms / 1000, timezone.utc).strftime(
|
||||
"%Y-%m-%d %H:%M:%S UTC"
|
||||
)
|
||||
|
||||
|
||||
def _database_endpoint(dsn: str) -> tuple[str, int]:
|
||||
parsed = urlparse(dsn)
|
||||
return parsed.hostname or "127.0.0.1", parsed.port or 8812
|
||||
|
||||
|
||||
def ensure_database(
|
||||
banbot_config: Path,
|
||||
dsn: str | None,
|
||||
) -> None:
|
||||
database_url = load_database_url(banbot_config, dsn)
|
||||
host, port = _database_endpoint(database_url)
|
||||
try:
|
||||
with socket.create_connection((host, port), timeout=1):
|
||||
return
|
||||
except OSError:
|
||||
pass
|
||||
if port != 8812:
|
||||
return
|
||||
|
||||
print("[latest-predict] QuestDB is unavailable; letting banbot start it", flush=True)
|
||||
start_ms = int(datetime.now(timezone.utc).timestamp() * 1000) - 8 * 60 * 60 * 1000
|
||||
end_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
|
||||
with tempfile.TemporaryDirectory(prefix="cbpred-db-bootstrap-") as output_dir:
|
||||
subprocess.run(
|
||||
[
|
||||
"go",
|
||||
"run",
|
||||
"./cmd/cbpred-export",
|
||||
"--config",
|
||||
str(banbot_config),
|
||||
"--data-dir",
|
||||
str(banbot_config.parent),
|
||||
"--pairs",
|
||||
"BTC/USDT:USDT",
|
||||
"--start-ms",
|
||||
str(start_ms),
|
||||
"--end-ms",
|
||||
str(end_ms),
|
||||
"--out-dir",
|
||||
output_dir,
|
||||
],
|
||||
cwd=REPO_ROOT,
|
||||
check=True,
|
||||
)
|
||||
|
||||
|
||||
def build_predictions(data_dir: Path, artifact: Path, output_dir: Path) -> dict[str, object]:
|
||||
manifest = json.loads((data_dir / "manifest.json").read_text())
|
||||
rows: list[dict[str, object]] = []
|
||||
total = len(manifest["symbols"])
|
||||
for index, item in enumerate(manifest["symbols"], 1):
|
||||
csv_path = data_dir / item["path"]
|
||||
frame = pd.read_csv(csv_path).sort_values("timestamp").drop_duplicates("timestamp")
|
||||
latest = frame.iloc[-1]
|
||||
result = predict_from_csv(artifact, csv_path, symbol=item["symbol"])
|
||||
eligible = bool(result["eligible_for_model"])
|
||||
rows.append(
|
||||
{
|
||||
"symbol": result["symbol"],
|
||||
"latest_bar_start_utc": _utc_label(int(result["timestamp"])),
|
||||
"close": result["close"],
|
||||
"quote_volume_usdt_8h": float(latest["quote"]),
|
||||
"atr_rank_200_pct": (
|
||||
result["volatility_rank_200"] * 100
|
||||
if result["volatility_rank_200"] is not None
|
||||
else None
|
||||
),
|
||||
"drawdown_point_pct": result["drawdown_point"] * 100 if eligible else None,
|
||||
"drawdown_cap_80pct_pct": (
|
||||
result["drawdown_upper80_q80"] * 100 if eligible else None
|
||||
),
|
||||
"price_level_reached_80pct": result["price_level_hit80"] if eligible else None,
|
||||
"price_level_cap_80pct": result["price_level_upper80"] if eligible else None,
|
||||
}
|
||||
)
|
||||
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)
|
||||
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.writeheader()
|
||||
writer.writerows(rows)
|
||||
|
||||
summary = {
|
||||
"model": str(artifact.resolve()),
|
||||
"symbols": len(rows),
|
||||
"eligible_predictions": sum(row["drawdown_point_pct"] is not None for row in rows),
|
||||
"latest_bar_start_utc": rows[0]["latest_bar_start_utc"] if rows else None,
|
||||
"csv": str(output_csv.resolve()),
|
||||
}
|
||||
(output_dir / "report.json").write_text(json.dumps(summary, indent=2, ensure_ascii=False))
|
||||
return summary
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description="Update banbot K-lines and predict with the current model")
|
||||
parser.add_argument("--banbot-dir", type=Path, default=Path("/home/ctyun/ban/banbot"))
|
||||
parser.add_argument(
|
||||
"--banbot-config",
|
||||
type=Path,
|
||||
default=Path("/home/ctyun/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("--output-dir", type=Path, default=REPO_ROOT / "predictions/latest_v1")
|
||||
parser.add_argument("--history-months", type=int, default=36)
|
||||
parser.add_argument("--holdout-months", type=int, default=10)
|
||||
parser.add_argument("--min-age-days", type=int, default=730)
|
||||
parser.add_argument("--chunk-size", type=int, default=100)
|
||||
parser.add_argument("--end", default=None, help="UTC date/ISO timestamp or milliseconds")
|
||||
parser.add_argument("--dsn", default=None)
|
||||
parser.add_argument("--db-type", choices=["auto", "questdb", "timescale"], default="auto")
|
||||
parser.add_argument("--exchange", default="binance")
|
||||
return parser
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = build_parser().parse_args()
|
||||
data_dir = args.data_dir.resolve()
|
||||
artifact = args.artifact.resolve()
|
||||
output_dir = args.output_dir.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) 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,
|
||||
)
|
||||
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))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user