1638 lines
64 KiB
Python
1638 lines
64 KiB
Python
from __future__ import annotations
|
|
|
|
import calendar
|
|
import json
|
|
import math
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
import joblib
|
|
import numpy as np
|
|
import pandas as pd
|
|
import yaml
|
|
from sklearn.ensemble import (
|
|
ExtraTreesRegressor,
|
|
HistGradientBoostingRegressor,
|
|
RandomForestRegressor,
|
|
)
|
|
from sklearn.metrics import mean_absolute_error, mean_squared_error
|
|
|
|
|
|
EIGHT_H_MS = 8 * 60 * 60 * 1000
|
|
HOUR_MS = 60 * 60 * 1000
|
|
WINDOW = 18
|
|
HORIZON = 6
|
|
VOLATILITY_RANK_WINDOW = 200
|
|
VOLATILITY_RANK_THRESHOLD = 0.0
|
|
LOWER_QUANTILE = 1 - 0.80
|
|
TARGET_QUANTILE = 0.80
|
|
FUTURE_LOW_QUANTILE = 0.20
|
|
TARGET_KIND = "signed_future_low_q20_return_atr_normalized"
|
|
CSV_COLUMNS = [
|
|
"timestamp",
|
|
"open",
|
|
"high",
|
|
"low",
|
|
"close",
|
|
"volume",
|
|
"quote",
|
|
"buy_volume",
|
|
"trade_num",
|
|
]
|
|
|
|
|
|
def _progress(message: str) -> None:
|
|
print(f"[cbpred] {message}", file=sys.stderr, flush=True)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Symbol:
|
|
sid: int
|
|
exchange: str
|
|
market: str
|
|
symbol: str
|
|
list_ms: int
|
|
delist_ms: int
|
|
combined: bool = False
|
|
|
|
|
|
def parse_time_ms(value: str | int | None) -> int:
|
|
if value is None:
|
|
return floor_to_8h(int(datetime.now(timezone.utc).timestamp() * 1000))
|
|
if isinstance(value, int):
|
|
return value
|
|
text = str(value).strip()
|
|
if text.isdigit():
|
|
number = int(text)
|
|
if number > 10**11:
|
|
return number
|
|
if len(text) == 8:
|
|
dt = datetime.strptime(text, "%Y%m%d").replace(tzinfo=timezone.utc)
|
|
return int(dt.timestamp() * 1000)
|
|
normalized = text.replace("Z", "+00:00")
|
|
dt = datetime.fromisoformat(normalized)
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return int(dt.timestamp() * 1000)
|
|
|
|
|
|
def floor_to_8h(value_ms: int) -> int:
|
|
return value_ms - value_ms % EIGHT_H_MS
|
|
|
|
|
|
def months_before(value_ms: int, months: int) -> int:
|
|
dt = datetime.fromtimestamp(value_ms / 1000, tz=timezone.utc)
|
|
month = dt.month - months
|
|
year = dt.year + (month - 1) // 12
|
|
month = (month - 1) % 12 + 1
|
|
day = min(dt.day, calendar.monthrange(year, month)[1])
|
|
result = dt.replace(year=year, month=month, day=day)
|
|
return int(result.timestamp() * 1000)
|
|
|
|
|
|
def load_database_url(config_path: Path | None, dsn: str | None) -> str:
|
|
if dsn:
|
|
return dsn
|
|
if not config_path:
|
|
raise ValueError("--dsn or --banbot-config is required")
|
|
data = yaml.safe_load(config_path.read_text()) or {}
|
|
url = ((data.get("database") or {}).get("url") or "").strip()
|
|
if not url:
|
|
raise ValueError(f"database.url is missing in {config_path}")
|
|
return url
|
|
|
|
|
|
def connect_database(dsn: str):
|
|
try:
|
|
import psycopg
|
|
except ImportError as exc:
|
|
raise RuntimeError("install dependencies with: python -m pip install -e .") from exc
|
|
return psycopg.connect(dsn, autocommit=True)
|
|
|
|
|
|
def detect_db_type(conn, requested: str = "auto") -> str:
|
|
if requested in {"questdb", "timescale"}:
|
|
return requested
|
|
with conn.cursor() as cur:
|
|
try:
|
|
cur.execute("SELECT 1 FROM exsymbol_q LIMIT 1")
|
|
return "questdb"
|
|
except Exception:
|
|
conn.rollback()
|
|
return "timescale"
|
|
|
|
|
|
def fetch_symbols(
|
|
conn,
|
|
db_type: str,
|
|
exchange: str,
|
|
eligible_at_ms: int,
|
|
min_age_days: int,
|
|
) -> tuple[list[Symbol], list[dict[str, Any]]]:
|
|
if db_type == "questdb":
|
|
sql = """
|
|
SELECT sid, exchange, market, symbol, list_ms, delist_ms, combined
|
|
FROM exsymbol_q
|
|
LATEST BY sid
|
|
WHERE exchange = %s AND coalesce(is_deleted, false) = false
|
|
ORDER BY sid
|
|
"""
|
|
else:
|
|
sql = """
|
|
SELECT id, exchange, market, symbol, list_ms, delist_ms, combined
|
|
FROM exsymbol
|
|
WHERE exchange = %s
|
|
ORDER BY id
|
|
"""
|
|
cutoff = eligible_at_ms - min_age_days * 24 * 60 * 60 * 1000
|
|
selected: list[Symbol] = []
|
|
excluded: list[dict[str, Any]] = []
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, (exchange,))
|
|
for row in cur.fetchall():
|
|
sid, exg, market, symbol, list_ms, delist_ms, combined = row
|
|
item = {
|
|
"sid": int(sid),
|
|
"exchange": str(exg),
|
|
"market": str(market),
|
|
"symbol": str(symbol),
|
|
"list_ms": int(list_ms or 0),
|
|
"delist_ms": int(delist_ms or 0),
|
|
"combined": bool(combined),
|
|
}
|
|
reason = None
|
|
if item["market"].lower() != "linear":
|
|
reason = "market_not_linear"
|
|
elif not is_usdt_linear_symbol(item["symbol"]):
|
|
reason = "not_usdt_settled"
|
|
elif item["combined"]:
|
|
reason = "combined_symbol"
|
|
elif item["list_ms"] <= 0:
|
|
reason = "missing_list_ms"
|
|
elif item["list_ms"] > cutoff:
|
|
reason = "listed_less_than_min_age"
|
|
elif item["delist_ms"] and item["delist_ms"] <= eligible_at_ms:
|
|
reason = "delisted_before_training_snapshot"
|
|
if reason:
|
|
item["reason"] = reason
|
|
excluded.append(item)
|
|
continue
|
|
selected.append(Symbol(**item))
|
|
return selected, excluded
|
|
|
|
|
|
def is_usdt_linear_symbol(symbol: str) -> bool:
|
|
parts = symbol.upper().split("/")
|
|
return len(parts) == 2 and parts[1] == "USDT:USDT"
|
|
|
|
|
|
def run_banbot_download(
|
|
symbols: list[Symbol],
|
|
banbot_dir: Path,
|
|
config_path: Path,
|
|
history_start_ms: int,
|
|
end_ms: int,
|
|
chunk_size: int = 100,
|
|
banbot_cmd: str = "go run .",
|
|
) -> None:
|
|
if not symbols:
|
|
raise ValueError("no eligible symbols to download")
|
|
base = shlex.split(banbot_cmd)
|
|
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)
|
|
|
|
|
|
def run_banbot_export(
|
|
symbols: list[Symbol],
|
|
banbot_dir: Path,
|
|
banbot_config: Path,
|
|
out_dir: Path,
|
|
history_start_ms: int,
|
|
end_ms: int,
|
|
banbot_export_cmd: str = "go run ./cmd/cbpred-export",
|
|
) -> None:
|
|
if not symbols:
|
|
raise ValueError("no eligible symbols to export")
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
pairs = ",".join(item.symbol for item in symbols)
|
|
command = shlex.split(banbot_export_cmd) + [
|
|
"--config",
|
|
str(banbot_config),
|
|
"--data-dir",
|
|
str(banbot_config.parent),
|
|
"--pairs",
|
|
pairs,
|
|
"--start-ms",
|
|
str(history_start_ms),
|
|
"--end-ms",
|
|
str(end_ms),
|
|
"--out-dir",
|
|
str(out_dir),
|
|
]
|
|
project_dir = Path(__file__).resolve().parents[1]
|
|
subprocess.run(command, cwd=project_dir, check=True)
|
|
|
|
|
|
def aggregate_8h(rows: Iterable[tuple[Any, ...]]) -> pd.DataFrame:
|
|
columns = CSV_COLUMNS
|
|
values = list(rows)
|
|
if not values:
|
|
return pd.DataFrame(columns=columns)
|
|
frame = pd.DataFrame(values, columns=columns)
|
|
for column in columns:
|
|
frame[column] = pd.to_numeric(frame[column], errors="coerce")
|
|
frame = frame.dropna(subset=["timestamp", "open", "high", "low", "close"])
|
|
frame = frame.sort_values("timestamp").drop_duplicates("timestamp", keep="last")
|
|
frame["timestamp"] = frame["timestamp"].astype(np.int64)
|
|
frame["quote"] = frame["quote"].fillna(0.0)
|
|
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
|
|
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):
|
|
continue
|
|
output.append(
|
|
{
|
|
"timestamp": int(bucket),
|
|
"open": float(group.iloc[0]["open"]),
|
|
"high": float(group["high"].max()),
|
|
"low": float(group["low"].min()),
|
|
"close": float(group.iloc[-1]["close"]),
|
|
"volume": float(group["volume"].sum()),
|
|
"quote": float(group["quote"].sum()),
|
|
"buy_volume": float(group["buy_volume"].sum()),
|
|
"trade_num": int(group["trade_num"].sum()),
|
|
}
|
|
)
|
|
return pd.DataFrame(output, columns=columns)
|
|
|
|
|
|
def symbol_filename(symbol: Symbol) -> str:
|
|
slug = symbol.symbol.replace("/", "_").replace(":", "_")
|
|
return f"{symbol.sid}_{slug}.csv"
|
|
|
|
|
|
def prepare_data(
|
|
out_dir: Path,
|
|
banbot_dir: Path,
|
|
banbot_config: Path,
|
|
dsn: str | None = None,
|
|
db_type: str = "auto",
|
|
exchange: str = "binance",
|
|
end_ms: int | None = None,
|
|
history_months: int = 36,
|
|
holdout_months: int = 10,
|
|
min_age_days: int = 730,
|
|
chunk_size: int = 100,
|
|
skip_download: bool = False,
|
|
banbot_cmd: str = "go run .",
|
|
banbot_export_cmd: str = "go run ./cmd/cbpred-export",
|
|
) -> dict[str, Any]:
|
|
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))
|
|
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)
|
|
conn = connect_database(database_url)
|
|
try:
|
|
actual_db_type = detect_db_type(conn, db_type)
|
|
symbols, excluded = fetch_symbols(
|
|
conn,
|
|
actual_db_type,
|
|
exchange,
|
|
training_snapshot_ms,
|
|
min_age_days,
|
|
)
|
|
finally:
|
|
conn.close()
|
|
if not skip_download:
|
|
run_banbot_download(
|
|
symbols,
|
|
banbot_dir,
|
|
banbot_config,
|
|
history_start_ms,
|
|
end_ms,
|
|
chunk_size,
|
|
banbot_cmd,
|
|
)
|
|
run_banbot_export(
|
|
symbols,
|
|
banbot_dir,
|
|
banbot_config,
|
|
raw_dir,
|
|
history_start_ms,
|
|
end_ms,
|
|
banbot_export_cmd,
|
|
)
|
|
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"})
|
|
continue
|
|
frame = pd.read_csv(path)
|
|
frame["timestamp"] = pd.to_numeric(frame["timestamp"], errors="coerce")
|
|
frame = frame.dropna(subset=["timestamp"]).sort_values("timestamp")
|
|
if symbol.delist_ms:
|
|
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"})
|
|
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",
|
|
"storage_timeframe": "banbot_orm",
|
|
"step_ms": EIGHT_H_MS,
|
|
"end_ms": end_ms,
|
|
"history_start_ms": history_start_ms,
|
|
"training_snapshot_ms": training_snapshot_ms,
|
|
"db_type": actual_db_type,
|
|
"exchange": exchange,
|
|
"symbols": manifest_symbols,
|
|
"excluded": excluded,
|
|
}
|
|
(out_dir / "manifest.json").write_text(json.dumps(manifest, indent=2, ensure_ascii=False))
|
|
return manifest
|
|
|
|
|
|
def load_manifest(data_dir: Path) -> dict[str, Any]:
|
|
path = data_dir / "manifest.json"
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"missing {path}; run prepare first")
|
|
return json.loads(path.read_text())
|
|
|
|
|
|
def load_raw_frames(data_dir: Path, manifest: dict[str, Any]) -> dict[str, pd.DataFrame]:
|
|
frames: dict[str, pd.DataFrame] = {}
|
|
for item in manifest.get("symbols", []):
|
|
path = data_dir / item["path"]
|
|
frame = pd.read_csv(path)
|
|
for column in CSV_COLUMNS:
|
|
frame[column] = pd.to_numeric(frame[column], errors="coerce")
|
|
frame = frame.dropna(subset=["timestamp", "open", "high", "low", "close"])
|
|
frame = frame.sort_values("timestamp").drop_duplicates("timestamp", keep="last").reset_index(drop=True)
|
|
frame["timestamp"] = frame["timestamp"].astype(np.int64)
|
|
delist_ms = int(item.get("delist_ms", 0) or 0)
|
|
if delist_ms:
|
|
frame = frame[frame["timestamp"] < delist_ms].reset_index(drop=True)
|
|
frame["quote"] = np.where(frame["quote"] > 0, frame["quote"], frame["volume"] * frame["close"])
|
|
frames[item["symbol"]] = frame
|
|
return frames
|
|
|
|
|
|
def compute_indicators(frame: pd.DataFrame) -> pd.DataFrame:
|
|
df = frame.copy()
|
|
close = df["close"].astype(float)
|
|
high = df["high"].astype(float)
|
|
low = df["low"].astype(float)
|
|
open_ = df["open"].astype(float)
|
|
previous = close.shift(1)
|
|
true_range = pd.concat(
|
|
[high - low, (high - previous).abs(), (low - previous).abs()], axis=1
|
|
).max(axis=1)
|
|
price_range = (high - low).clip(lower=1e-12)
|
|
log_return = np.log(close / previous.replace(0, np.nan))
|
|
df["body_sign"] = np.sign(close - open_)
|
|
df["body_range"] = (close - open_) / price_range
|
|
df["body_abs_range"] = (close - open_).abs() / price_range
|
|
df["upper_wick"] = (high - pd.concat([open_, close], axis=1).max(axis=1)) / price_range
|
|
df["lower_wick"] = (pd.concat([open_, close], axis=1).min(axis=1) - low) / price_range
|
|
df["close_location"] = 2 * (close - low) / price_range - 1
|
|
df["range_pct"] = price_range / close.replace(0, np.nan)
|
|
df["log_return"] = log_return.clip(-2, 2)
|
|
df["atr_pct"] = true_range.ewm(alpha=1 / 14, adjust=False, min_periods=14).mean() / close
|
|
df["atr_rank_200"] = df["atr_pct"].rolling(
|
|
VOLATILITY_RANK_WINDOW, min_periods=VOLATILITY_RANK_WINDOW
|
|
).rank(method="average", pct=True)
|
|
delta = close.diff()
|
|
gain = delta.clip(lower=0).ewm(alpha=1 / 14, adjust=False, min_periods=14).mean()
|
|
loss = (-delta.clip(upper=0)).ewm(alpha=1 / 14, adjust=False, min_periods=14).mean()
|
|
rs = gain / loss.replace(0, np.nan)
|
|
rsi = (100 - 100 / (1 + rs)) / 100
|
|
df["rsi_norm"] = rsi.where(loss > 0, np.where(gain > 0, 1.0, 0.5))
|
|
sma20 = close.rolling(20, min_periods=20).mean()
|
|
std20 = close.rolling(20, min_periods=20).std(ddof=0)
|
|
df["bb_position"] = ((close - sma20) / (2 * std20.replace(0, np.nan))).clip(-5, 5)
|
|
df["bb_width"] = (4 * std20 / sma20.replace(0, np.nan)).clip(0, 5)
|
|
ema9 = close.ewm(span=9, adjust=False, min_periods=9).mean()
|
|
ema21 = close.ewm(span=21, adjust=False, min_periods=21).mean()
|
|
df["ema9_gap"] = (close / ema9 - 1).clip(-2, 2)
|
|
df["ema21_gap"] = (close / ema21 - 1).clip(-2, 2)
|
|
df["trend_3"] = (close / close.shift(3) - 1).clip(-2, 2)
|
|
df["trend_9"] = (close / close.shift(9) - 1).clip(-2, 2)
|
|
df["volatility_18"] = log_return.rolling(18, min_periods=18).std(ddof=0).clip(0, 2)
|
|
df["drawdown_18"] = (close / close.rolling(18, min_periods=18).max() - 1).clip(-2, 0)
|
|
df["quote_log"] = np.log1p(df["quote"].clip(lower=0))
|
|
return df
|
|
|
|
|
|
def asset_profile(indicators: pd.DataFrame, train_end_ms: int) -> dict[str, float] | None:
|
|
history = indicators[indicators["timestamp"] < train_end_ms]
|
|
atr = history["atr_pct"].dropna().to_numpy(dtype=float)
|
|
volume = history["quote_log"].replace([np.inf, -np.inf], np.nan).dropna().to_numpy(dtype=float)
|
|
if len(atr) < 20 or len(volume) < 20:
|
|
return None
|
|
atr_top = np.sort(atr)[-max(1, math.ceil(len(atr) * 0.20)) :]
|
|
volume_top = np.sort(volume)[-max(1, math.ceil(len(volume) * 0.20)) :]
|
|
return {
|
|
"asset_atr_price_top20_median": float(np.median(atr_top)),
|
|
"asset_quote_volume_top20_median_log": float(np.median(volume_top)),
|
|
}
|
|
|
|
|
|
def fit_asset_scaler(
|
|
profiles: dict[str, dict[str, float]],
|
|
symbols: Iterable[str] | None = None,
|
|
) -> dict[str, dict[str, float]]:
|
|
names = ["asset_atr_price_top20_median", "asset_quote_volume_top20_median_log"]
|
|
selected = set(symbols) if symbols is not None else set(profiles)
|
|
result: dict[str, dict[str, float]] = {}
|
|
for name in names:
|
|
values = np.array([profiles[symbol][name] for symbol in selected if symbol in profiles], dtype=float)
|
|
if len(values) == 0:
|
|
raise ValueError("no asset profiles available for scaler")
|
|
result[name] = {"mean": float(values.mean()), "std": float(max(values.std(ddof=0), 1e-9))}
|
|
return result
|
|
|
|
|
|
def scale_asset_profile(profile: dict[str, float], scaler: dict[str, dict[str, float]]) -> dict[str, float]:
|
|
return {
|
|
name: (profile[name] - values["mean"]) / values["std"]
|
|
for name, values in scaler.items()
|
|
}
|
|
|
|
|
|
_BASE_FEATURE_COLUMNS = (
|
|
"body_sign",
|
|
"body_range",
|
|
"body_abs_range",
|
|
"upper_wick",
|
|
"lower_wick",
|
|
"close_location",
|
|
"range_pct",
|
|
"log_return",
|
|
)
|
|
_INDICATOR_FEATURE_COLUMNS = (
|
|
"atr_pct",
|
|
"atr_rank_200",
|
|
"rsi_norm",
|
|
"bb_position",
|
|
"bb_width",
|
|
"ema9_gap",
|
|
"ema21_gap",
|
|
"trend_3",
|
|
"trend_9",
|
|
"volatility_18",
|
|
"drawdown_18",
|
|
)
|
|
_ASSET_FEATURE_NAMES = (
|
|
"asset_atr_price_top20_median",
|
|
"asset_quote_volume_top20_median_log",
|
|
)
|
|
_SUMMARY_FEATURE_NAMES = (
|
|
"summary_return_18",
|
|
"summary_return_3",
|
|
"summary_return_6",
|
|
"summary_range_mean",
|
|
"summary_range_std",
|
|
"summary_range_max_rel_mean",
|
|
"summary_up_ratio",
|
|
"summary_down_ratio",
|
|
"summary_close_position",
|
|
"summary_volume_trend",
|
|
"summary_return_mean",
|
|
"summary_return_std",
|
|
"summary_rsi_norm",
|
|
"summary_bb_position",
|
|
"summary_bb_width",
|
|
)
|
|
|
|
|
|
def _window_feature_names(stage: int) -> list[str]:
|
|
names: list[str] = []
|
|
for lag in range(WINDOW - 1, -1, -1):
|
|
names.extend(f"bar_{lag:02d}_{name}" for name in _BASE_FEATURE_COLUMNS)
|
|
names.extend(
|
|
[
|
|
f"bar_{lag:02d}_quote_volume_rank",
|
|
f"bar_{lag:02d}_quote_volume_log_rel_median",
|
|
]
|
|
)
|
|
if stage >= 2:
|
|
names.extend(f"bar_{lag:02d}_{name}" for name in _INDICATOR_FEATURE_COLUMNS)
|
|
names.extend(_ASSET_FEATURE_NAMES)
|
|
if stage >= 3:
|
|
names.extend(_SUMMARY_FEATURE_NAMES)
|
|
return names
|
|
|
|
|
|
def _rank_windows(values: np.ndarray) -> np.ndarray:
|
|
"""Return pandas-compatible average percentile ranks for each row."""
|
|
width = values.shape[1]
|
|
order = np.argsort(values, axis=1, kind="mergesort")
|
|
sorted_values = np.take_along_axis(values, order, axis=1)
|
|
same_previous = np.zeros(sorted_values.shape, dtype=bool)
|
|
same_previous[:, 1:] = sorted_values[:, 1:] == sorted_values[:, :-1]
|
|
group_start = ~same_previous
|
|
positions = np.arange(1, width + 1, dtype=float)[None, :]
|
|
start_ranks = np.where(group_start, positions, 0.0)
|
|
start_ranks = np.maximum.accumulate(start_ranks, axis=1)
|
|
|
|
same_next = np.zeros(sorted_values.shape, dtype=bool)
|
|
same_next[:, :-1] = sorted_values[:, :-1] == sorted_values[:, 1:]
|
|
group_end = ~same_next
|
|
end_ranks = np.where(group_end, positions, float(width + 1))
|
|
end_ranks = np.minimum.accumulate(end_ranks[:, ::-1], axis=1)[:, ::-1]
|
|
sorted_ranks = (start_ranks + end_ranks) / (2.0 * width)
|
|
ranks = np.empty_like(sorted_ranks)
|
|
np.put_along_axis(ranks, order, sorted_ranks, axis=1)
|
|
return ranks
|
|
|
|
|
|
def _vectorized_window_features(
|
|
frame: pd.DataFrame,
|
|
stage: int,
|
|
anchor_indices: np.ndarray,
|
|
asset: dict[str, float],
|
|
) -> np.ndarray:
|
|
starts = anchor_indices - (WINDOW - 1)
|
|
quote = frame["quote_log"].to_numpy(dtype=float)
|
|
quote_windows = np.lib.stride_tricks.sliding_window_view(quote, WINDOW)[starts]
|
|
volume_rank = _rank_windows(quote_windows)
|
|
volume_median = np.median(quote_windows, axis=1, keepdims=True)
|
|
|
|
columns = list(_BASE_FEATURE_COLUMNS)
|
|
if stage >= 2:
|
|
columns.extend(_INDICATOR_FEATURE_COLUMNS)
|
|
values = frame[columns].to_numpy(dtype=float)
|
|
windows = np.lib.stride_tricks.sliding_window_view(values, WINDOW, axis=0)
|
|
windows = windows.transpose(0, 2, 1)[starts]
|
|
parts = [windows[:, :, : len(_BASE_FEATURE_COLUMNS)]]
|
|
parts.extend(
|
|
[
|
|
volume_rank[:, :, None],
|
|
(quote_windows - volume_median)[:, :, None],
|
|
]
|
|
)
|
|
if stage >= 2:
|
|
parts.append(windows[:, :, len(_BASE_FEATURE_COLUMNS) :])
|
|
feature_matrix = np.concatenate(parts, axis=2).reshape(len(anchor_indices), -1)
|
|
|
|
asset_values = np.array([asset[name] for name in _ASSET_FEATURE_NAMES], dtype=float)
|
|
feature_matrix = np.column_stack(
|
|
[feature_matrix, np.repeat(asset_values[None, :], len(anchor_indices), axis=0)]
|
|
)
|
|
if stage < 3:
|
|
return feature_matrix
|
|
|
|
close = frame["close"].to_numpy(dtype=float)
|
|
high = frame["high"].to_numpy(dtype=float)
|
|
low = frame["low"].to_numpy(dtype=float)
|
|
log_return = frame["log_return"].to_numpy(dtype=float)
|
|
body_sign = frame["body_sign"].to_numpy(dtype=float)
|
|
range_pct = frame["range_pct"].to_numpy(dtype=float)
|
|
close_windows = np.lib.stride_tricks.sliding_window_view(close, WINDOW)[starts]
|
|
high_windows = np.lib.stride_tricks.sliding_window_view(high, WINDOW)[starts]
|
|
low_windows = np.lib.stride_tricks.sliding_window_view(low, WINDOW)[starts]
|
|
return_windows = np.lib.stride_tricks.sliding_window_view(log_return, WINDOW)[starts]
|
|
sign_windows = np.lib.stride_tricks.sliding_window_view(body_sign, WINDOW)[starts]
|
|
range_windows = np.lib.stride_tricks.sliding_window_view(range_pct, WINDOW)[starts]
|
|
x = np.arange(WINDOW, dtype=float)
|
|
x_centered = x - x.mean()
|
|
x_sum_squares = float(np.sum(x_centered * x_centered))
|
|
range_mean = range_windows.mean(axis=1)
|
|
summaries = np.column_stack(
|
|
[
|
|
np.log(close_windows[:, -1] / close_windows[:, 0]),
|
|
np.log(close_windows[:, -1] / close_windows[:, -4]),
|
|
np.log(close_windows[:, -1] / close_windows[:, -7]),
|
|
range_mean,
|
|
range_windows.std(axis=1),
|
|
range_windows[:, -1] / np.maximum(range_mean, 1e-9),
|
|
np.mean(sign_windows > 0, axis=1),
|
|
np.mean(sign_windows < 0, axis=1),
|
|
(close_windows[:, -1] - low_windows.min(axis=1))
|
|
/ np.maximum(high_windows.max(axis=1) - low_windows.min(axis=1), 1e-9),
|
|
((quote_windows - quote_windows.mean(axis=1, keepdims=True)) * x_centered).sum(axis=1)
|
|
/ x_sum_squares,
|
|
np.nanmean(return_windows, axis=1),
|
|
np.nanstd(return_windows, axis=1),
|
|
frame["rsi_norm"].to_numpy(dtype=float)[anchor_indices],
|
|
frame["bb_position"].to_numpy(dtype=float)[anchor_indices],
|
|
frame["bb_width"].to_numpy(dtype=float)[anchor_indices],
|
|
]
|
|
)
|
|
return np.column_stack([feature_matrix, summaries])
|
|
|
|
|
|
def _window_features(window: pd.DataFrame, stage: int, asset: dict[str, float]) -> dict[str, float]:
|
|
result: dict[str, float] = {}
|
|
volume_rank = window["quote_log"].rank(method="average", pct=True).to_numpy(dtype=float)
|
|
volume_median = float(window["quote_log"].median())
|
|
base_values = window[list(_BASE_FEATURE_COLUMNS)].to_numpy(dtype=float)
|
|
indicator_values = (
|
|
window[list(_INDICATOR_FEATURE_COLUMNS)].to_numpy(dtype=float) if stage >= 2 else None
|
|
)
|
|
quote_values = window["quote_log"].to_numpy(dtype=float)
|
|
for position in range(len(window)):
|
|
lag = WINDOW - 1 - position
|
|
for column_index, name in enumerate(_BASE_FEATURE_COLUMNS):
|
|
result[f"bar_{lag:02d}_{name}"] = float(base_values[position, column_index])
|
|
result[f"bar_{lag:02d}_quote_volume_rank"] = float(volume_rank[position])
|
|
result[f"bar_{lag:02d}_quote_volume_log_rel_median"] = float(quote_values[position] - volume_median)
|
|
if indicator_values is not None:
|
|
for column_index, name in enumerate(_INDICATOR_FEATURE_COLUMNS):
|
|
result[f"bar_{lag:02d}_{name}"] = float(indicator_values[position, column_index])
|
|
result.update(asset)
|
|
if stage >= 3:
|
|
close = window["close"].to_numpy(dtype=float)
|
|
returns = window["log_return"].to_numpy(dtype=float)
|
|
ranges = window["range_pct"].to_numpy(dtype=float)
|
|
signs = window["body_sign"].to_numpy(dtype=float)
|
|
result.update(
|
|
{
|
|
"summary_return_18": float(np.log(close[-1] / close[0])),
|
|
"summary_return_3": float(np.log(close[-1] / close[-4])),
|
|
"summary_return_6": float(np.log(close[-1] / close[-7])),
|
|
"summary_range_mean": float(np.mean(ranges)),
|
|
"summary_range_std": float(np.std(ranges)),
|
|
"summary_range_max_rel_mean": float(ranges[-1] / max(np.mean(ranges), 1e-9)),
|
|
"summary_up_ratio": float(np.mean(signs > 0)),
|
|
"summary_down_ratio": float(np.mean(signs < 0)),
|
|
"summary_close_position": float(
|
|
(close[-1] - window["low"].min())
|
|
/ max(window["high"].max() - window["low"].min(), 1e-9)
|
|
),
|
|
"summary_volume_trend": float(
|
|
np.polyfit(np.arange(WINDOW), window["quote_log"].to_numpy(dtype=float), 1)[0]
|
|
),
|
|
"summary_return_mean": float(np.nanmean(returns)),
|
|
"summary_return_std": float(np.nanstd(returns)),
|
|
"summary_rsi_norm": float(window["rsi_norm"].iloc[-1]),
|
|
"summary_bb_position": float(window["bb_position"].iloc[-1]),
|
|
"summary_bb_width": float(window["bb_width"].iloc[-1]),
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def build_feature_dataset(
|
|
frames: dict[str, pd.DataFrame],
|
|
stage: int,
|
|
test_start_ms: int,
|
|
end_ms: int,
|
|
asset_scaler: dict[str, dict[str, float]],
|
|
train_symbols: Iterable[str] | None = None,
|
|
validation_symbols: Iterable[str] | None = None,
|
|
test_symbols: Iterable[str] | None = None,
|
|
validation_end_ms: int | None = None,
|
|
calibration_end_ms: int | None = None,
|
|
volatility_rank_threshold: float = VOLATILITY_RANK_THRESHOLD,
|
|
) -> 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")
|
|
indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()}
|
|
raw_profiles = {
|
|
symbol: asset_profile(frame, test_start_ms)
|
|
for symbol, frame in indicators.items()
|
|
}
|
|
profiles = {symbol: profile for symbol, profile in raw_profiles.items() if profile is not None}
|
|
if not profiles:
|
|
raise ValueError("no symbol has enough pre-test history for asset profiles")
|
|
scaled_profiles = {
|
|
symbol: scale_asset_profile(profile, asset_scaler)
|
|
for symbol, profile in profiles.items()
|
|
}
|
|
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
|
|
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 [])
|
|
test_set = set(test_symbols or [])
|
|
if explicit_splits and (validation_end_ms is None or calibration_end_ms is None):
|
|
raise ValueError("validation_end_ms and calibration_end_ms are required for explicit splits")
|
|
for symbol, frame in indicators.items():
|
|
asset = scaled_profiles.get(symbol)
|
|
if asset is None:
|
|
continue
|
|
frame = frame.reset_index(drop=True)
|
|
timestamps = frame["timestamp"].to_numpy(dtype=np.int64)
|
|
if len(frame) < WINDOW + HORIZON:
|
|
continue
|
|
anchor_indices = np.arange(WINDOW - 1, len(frame) - HORIZON, dtype=np.int64)
|
|
continuity = np.convolve(
|
|
(np.diff(timestamps) == step).astype(np.int8),
|
|
np.ones(WINDOW + HORIZON - 1, dtype=np.int8),
|
|
mode="valid",
|
|
) == WINDOW + HORIZON - 1
|
|
anchor_ms = timestamps[anchor_indices]
|
|
future_end_ms = timestamps[anchor_indices + HORIZON] + step
|
|
split_labels = np.full(len(anchor_indices), "", dtype=object)
|
|
if explicit_splits:
|
|
if symbol in train_set:
|
|
mask = (anchor_ms < test_start_ms) & (future_end_ms <= test_start_ms)
|
|
split_labels[mask] = "train"
|
|
if symbol in validation_set:
|
|
mask = (
|
|
(test_start_ms <= anchor_ms)
|
|
& (anchor_ms < int(validation_end_ms))
|
|
& (future_end_ms <= int(validation_end_ms))
|
|
)
|
|
split_labels[mask] = "validation"
|
|
mask = (
|
|
(int(validation_end_ms) <= anchor_ms)
|
|
& (anchor_ms < int(calibration_end_ms))
|
|
& (future_end_ms <= int(calibration_end_ms))
|
|
)
|
|
split_labels[mask] = "calibration"
|
|
if symbol in test_set:
|
|
mask = (
|
|
(int(calibration_end_ms) <= anchor_ms)
|
|
& (anchor_ms < end_ms)
|
|
& (future_end_ms <= end_ms)
|
|
)
|
|
split_labels[mask] = "test"
|
|
else:
|
|
mask = (anchor_ms < test_start_ms) & (future_end_ms <= test_start_ms)
|
|
split_labels[mask] = "train"
|
|
mask = (
|
|
(test_start_ms <= anchor_ms)
|
|
& (anchor_ms < end_ms)
|
|
& (future_end_ms <= end_ms)
|
|
)
|
|
split_labels[mask] = "recent"
|
|
candidate = continuity & (split_labels != "")
|
|
for split_name in candidate_counts:
|
|
candidate_counts[split_name] += int(np.count_nonzero(candidate & (split_labels == split_name)))
|
|
volatility_rank = frame["atr_rank_200"].to_numpy(dtype=float)[anchor_indices]
|
|
if volatility_rank_threshold == 0.0:
|
|
selected = candidate
|
|
else:
|
|
selected = candidate & np.isfinite(volatility_rank) & (volatility_rank >= volatility_rank_threshold)
|
|
for split_name in selected_counts:
|
|
selected_counts[split_name] += int(np.count_nonzero(selected & (split_labels == split_name)))
|
|
if not np.any(selected):
|
|
continue
|
|
selected_indices = anchor_indices[selected]
|
|
feature_matrix = _vectorized_window_features(frame, stage, selected_indices, asset)
|
|
low = frame["low"].to_numpy(dtype=float)
|
|
close = frame["close"].to_numpy(dtype=float)[selected_indices]
|
|
future_lows = np.column_stack(
|
|
[low[selected_indices + offset] for offset in range(1, HORIZON + 1)]
|
|
)
|
|
future_low_reference = np.quantile(future_lows, FUTURE_LOW_QUANTILE, axis=1)
|
|
future_low_return = future_low_reference / np.maximum(close, 1e-12) - 1.0
|
|
target_scale = max(
|
|
float(profiles[symbol]["asset_atr_price_top20_median"]),
|
|
1e-9,
|
|
)
|
|
target = future_low_return / target_scale
|
|
sample_atr_pct = frame["atr_pct"].to_numpy(dtype=float)[selected_indices]
|
|
finite = (
|
|
np.isfinite(target)
|
|
& np.isfinite(target_scale)
|
|
& np.isfinite(sample_atr_pct)
|
|
& np.isfinite(feature_matrix).all(axis=1)
|
|
)
|
|
if not np.any(finite):
|
|
continue
|
|
feature_names = _window_feature_names(stage)
|
|
block = pd.DataFrame(feature_matrix[finite], columns=feature_names)
|
|
block["symbol"] = symbol
|
|
block["timestamp"] = anchor_ms[selected][finite]
|
|
block["target"] = target[finite]
|
|
block["target_scale"] = target_scale
|
|
block["split"] = split_labels[selected][finite]
|
|
block["sample_atr_pct"] = sample_atr_pct[finite]
|
|
rows.append(block)
|
|
dataset = pd.concat(rows, ignore_index=True) if rows else pd.DataFrame()
|
|
if dataset.empty:
|
|
raise ValueError("feature dataset is empty")
|
|
dataset.attrs["volatility_filter"] = {
|
|
"window": VOLATILITY_RANK_WINDOW,
|
|
"threshold": volatility_rank_threshold,
|
|
"candidate_counts": candidate_counts,
|
|
"selected_counts": selected_counts,
|
|
}
|
|
feature_names = _window_feature_names(stage)
|
|
return dataset, feature_names, profiles
|
|
|
|
|
|
def split_holdout_symbols(symbols: Iterable[str], seed: int = 42, holdout_ratio: float = 0.40) -> tuple[list[str], list[str], list[str]]:
|
|
available = sorted(set(symbols))
|
|
if len(available) < 3:
|
|
raise ValueError("at least 3 symbols are required for train/validation/test by symbol")
|
|
rng = np.random.default_rng(seed)
|
|
shuffled = list(np.array(available)[rng.permutation(len(available))])
|
|
holdout_count = max(2, int(math.ceil(len(shuffled) * holdout_ratio)))
|
|
holdout_count = min(holdout_count, len(shuffled) - 1)
|
|
validation_count = max(1, holdout_count // 2)
|
|
validation = sorted(shuffled[:validation_count])
|
|
test = sorted(shuffled[validation_count:holdout_count])
|
|
train_symbols = sorted(shuffled[holdout_count:])
|
|
return train_symbols, validation, test
|
|
|
|
|
|
def recent_split_boundaries(start_ms: int, end_ms: int) -> 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))
|
|
return validation_end, calibration_end
|
|
|
|
|
|
def _candidate_specs(random_state: int, n_estimators: int, n_jobs: int) -> list[dict[str, Any]]:
|
|
tree_common = dict(
|
|
n_estimators=n_estimators,
|
|
max_features=0.7,
|
|
min_samples_leaf=8,
|
|
random_state=random_state,
|
|
n_jobs=n_jobs,
|
|
verbose=1,
|
|
)
|
|
return [
|
|
{
|
|
"name": "random_forest_squared_error",
|
|
"kind": "point_plus_gbdt_quantiles",
|
|
"point_factory": lambda: RandomForestRegressor(criterion="squared_error", **tree_common),
|
|
},
|
|
{
|
|
"name": "extra_trees_squared_error",
|
|
"kind": "point_plus_gbdt_quantiles",
|
|
"point_factory": lambda: ExtraTreesRegressor(criterion="squared_error", **tree_common),
|
|
},
|
|
]
|
|
|
|
|
|
def _quantile_model(
|
|
alpha: float, random_state: int, n_estimators: int
|
|
) -> HistGradientBoostingRegressor:
|
|
return HistGradientBoostingRegressor(
|
|
quantile=alpha,
|
|
max_iter=max(1, n_estimators),
|
|
learning_rate=0.04,
|
|
max_depth=2,
|
|
min_samples_leaf=8,
|
|
random_state=random_state,
|
|
verbose=1,
|
|
)
|
|
|
|
|
|
def predict_quantiles(
|
|
point_model,
|
|
lower_model,
|
|
upper_model,
|
|
features: pd.DataFrame,
|
|
model_kind: str,
|
|
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
|
|
if model_kind != "point_plus_gbdt_quantiles":
|
|
raise ValueError(f"unsupported model kind: {model_kind}")
|
|
point = np.asarray(point_model.predict(features), dtype=float)
|
|
lower = np.asarray(lower_model.predict(features), dtype=float)
|
|
upper = np.asarray(upper_model.predict(features), dtype=float)
|
|
return point, lower, upper
|
|
|
|
|
|
def pinball_loss(y_true: np.ndarray, prediction: np.ndarray, quantile: float) -> float:
|
|
error = y_true - prediction
|
|
return float(np.mean(np.maximum(quantile * error, (quantile - 1) * error)))
|
|
|
|
|
|
def evaluate_predictions(
|
|
y_true: np.ndarray,
|
|
point: np.ndarray,
|
|
lower: np.ndarray,
|
|
upper: np.ndarray,
|
|
target_scale: np.ndarray | None = None,
|
|
) -> dict[str, float]:
|
|
scale = np.ones_like(y_true) if target_scale is None else np.asarray(target_scale, dtype=float)
|
|
actual_y = y_true * scale
|
|
actual_point = point * scale
|
|
actual_lower = lower * scale
|
|
actual_upper = upper * scale
|
|
q20_coverage = float(np.mean(y_true <= lower))
|
|
q80_coverage = float(np.mean(y_true <= upper))
|
|
return {
|
|
"n": int(len(y_true)),
|
|
"mae_point": float(mean_absolute_error(actual_y, actual_point)),
|
|
"rmse_point": float(math.sqrt(mean_squared_error(actual_y, actual_point))),
|
|
"mae_q20": float(mean_absolute_error(actual_y, actual_lower)),
|
|
"rmse_q20": float(math.sqrt(mean_squared_error(actual_y, actual_lower))),
|
|
"mae_q80": float(mean_absolute_error(actual_y, actual_upper)),
|
|
"rmse_q80": float(math.sqrt(mean_squared_error(actual_y, actual_upper))),
|
|
"pinball_q20": pinball_loss(actual_y, actual_lower, LOWER_QUANTILE),
|
|
"pinball_q80": pinball_loss(actual_y, actual_upper, TARGET_QUANTILE),
|
|
"q20_coverage": q20_coverage,
|
|
"q80_coverage": q80_coverage,
|
|
"q20_hit_rate": float(np.mean(y_true >= lower)),
|
|
"q20_calibration_gap": float(abs(q20_coverage - LOWER_QUANTILE)),
|
|
"q80_calibration_gap": float(abs(q80_coverage - TARGET_QUANTILE)),
|
|
"quantile_cross_rate": float(np.mean(lower > upper)),
|
|
}
|
|
|
|
|
|
def _fit_stage(
|
|
dataset: pd.DataFrame,
|
|
feature_names: list[str],
|
|
validation_symbols: list[str],
|
|
stage: int,
|
|
random_state: int,
|
|
n_estimators: int,
|
|
n_jobs: int,
|
|
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
train = dataset[dataset["split"] == "train"]
|
|
validation = dataset[(dataset["split"] == "validation") & dataset["symbol"].isin(validation_symbols)]
|
|
if train.empty or validation.empty:
|
|
raise ValueError("train or validation samples are empty")
|
|
train_x = train[feature_names]
|
|
train_y = train["target"].to_numpy(dtype=float)
|
|
symbol_counts = train["symbol"].value_counts()
|
|
train_weight = (
|
|
1.0 / train["symbol"].map(symbol_counts).to_numpy(dtype=float)
|
|
)
|
|
train_weight /= train_weight.mean()
|
|
validation_x = validation[feature_names]
|
|
validation_y = validation["target"].to_numpy(dtype=float)
|
|
validation_scale = validation["target_scale"].to_numpy(dtype=float)
|
|
results: list[dict[str, Any]] = []
|
|
best: dict[str, Any] | None = None
|
|
# Quantile models depend only on the feature stage, not on the point-model
|
|
# candidate. Fit them once per stage so candidate comparison stays cheap.
|
|
_progress(f"stage {stage}: fitting q20 quantile model")
|
|
lower_model = _quantile_model(LOWER_QUANTILE, random_state, n_estimators)
|
|
upper_model = _quantile_model(TARGET_QUANTILE, random_state, n_estimators)
|
|
lower_model.fit(train_x, train_y)
|
|
_progress(f"stage {stage}: q20 quantile model ready")
|
|
_progress(f"stage {stage}: fitting q80 quantile model")
|
|
upper_model.fit(train_x, train_y)
|
|
_progress(f"stage {stage}: quantile models ready")
|
|
for spec in _candidate_specs(random_state, n_estimators, n_jobs):
|
|
_progress(f"stage {stage}: fitting {spec['name']} point model")
|
|
point_model = spec["point_factory"]()
|
|
point_model.fit(train_x, train_y, sample_weight=train_weight)
|
|
point, lower, upper = predict_quantiles(
|
|
point_model, lower_model, upper_model, validation_x, spec["kind"]
|
|
)
|
|
metrics = evaluate_predictions(validation_y, point, lower, upper, validation_scale)
|
|
selection_score = (
|
|
metrics["pinball_q20"]
|
|
+ metrics["pinball_q80"]
|
|
+ 0.20 * (metrics["q20_calibration_gap"] + metrics["q80_calibration_gap"])
|
|
+ 0.05 * metrics["mae_point"]
|
|
)
|
|
record = {"model": spec["name"], "selection_score": float(selection_score), **metrics}
|
|
results.append(record)
|
|
_progress(
|
|
f"stage {stage}: {spec['name']} validation score={selection_score:.6f}"
|
|
)
|
|
if best is None or selection_score < best["selection_score"]:
|
|
best = {
|
|
"spec": spec,
|
|
"point_model": point_model,
|
|
"lower_model": lower_model,
|
|
"upper_model": upper_model,
|
|
"validation_metrics": metrics,
|
|
"selection_score": float(selection_score),
|
|
}
|
|
if best is None:
|
|
raise RuntimeError("no model candidate was trained")
|
|
best["feature_names"] = feature_names
|
|
return best, results
|
|
|
|
|
|
def _calibration_offsets(best: dict[str, Any], calibration: pd.DataFrame) -> dict[str, float]:
|
|
features = calibration[best["feature_names"]]
|
|
y = calibration["target"].to_numpy(dtype=float)
|
|
_, lower, upper = predict_quantiles(
|
|
best["point_model"],
|
|
best["lower_model"],
|
|
best["upper_model"],
|
|
features,
|
|
best["spec"]["kind"],
|
|
)
|
|
return {
|
|
"q20_offset": float(np.quantile(y - lower, LOWER_QUANTILE)),
|
|
"q80_offset": float(np.quantile(y - upper, TARGET_QUANTILE)),
|
|
}
|
|
|
|
|
|
def apply_calibration(
|
|
lower: np.ndarray,
|
|
upper: np.ndarray,
|
|
offsets: dict[str, float],
|
|
) -> tuple[np.ndarray, np.ndarray]:
|
|
lower = lower + float(offsets.get("q20_offset", 0.0))
|
|
upper = upper + float(offsets.get("q80_offset", 0.0))
|
|
return np.minimum(lower, upper), np.maximum(lower, upper)
|
|
|
|
|
|
def _group_metrics(
|
|
dataset: pd.DataFrame,
|
|
y_true: np.ndarray,
|
|
point: np.ndarray,
|
|
lower: np.ndarray,
|
|
upper: np.ndarray,
|
|
column: str,
|
|
) -> dict[str, dict[str, float]]:
|
|
result: dict[str, dict[str, float]] = {}
|
|
for group, group_frame in dataset.reset_index(drop=True).groupby(column, sort=True):
|
|
indices = group_frame.index.to_numpy(dtype=np.int64)
|
|
result[str(group)] = evaluate_predictions(
|
|
y_true[indices], point[indices], lower[indices], upper[indices]
|
|
)
|
|
return result
|
|
|
|
|
|
def _stratified_metrics(
|
|
dataset: pd.DataFrame,
|
|
y_true: np.ndarray,
|
|
point: np.ndarray,
|
|
lower: np.ndarray,
|
|
upper: np.ndarray,
|
|
) -> dict[str, dict[str, dict[str, float]]]:
|
|
frame = dataset.reset_index(drop=True).copy()
|
|
frame["month"] = pd.to_datetime(frame["timestamp"], unit="ms", utc=True).dt.strftime("%Y-%m")
|
|
result: dict[str, dict[str, dict[str, float]]] = {
|
|
"by_symbol": _group_metrics(frame, y_true, point, lower, upper, "symbol"),
|
|
"by_month": _group_metrics(frame, y_true, point, lower, upper, "month"),
|
|
}
|
|
if frame["sample_atr_pct"].nunique() >= 3:
|
|
frame["volatility_bucket"] = pd.qcut(
|
|
frame["sample_atr_pct"], q=3, labels=["low", "medium", "high"], duplicates="drop"
|
|
).astype(str)
|
|
result["by_volatility"] = _group_metrics(
|
|
frame, y_true, point, lower, upper, "volatility_bucket"
|
|
)
|
|
else:
|
|
result["by_volatility"] = _group_metrics(
|
|
frame.assign(volatility_bucket="all"), y_true, point, lower, upper, "volatility_bucket"
|
|
)
|
|
return result
|
|
|
|
|
|
def _block_bootstrap_ci(
|
|
dataset: pd.DataFrame,
|
|
y_true: np.ndarray,
|
|
point: np.ndarray,
|
|
lower: np.ndarray,
|
|
upper: np.ndarray,
|
|
seed: int,
|
|
repetitions: int = 100,
|
|
block_size: int = 18,
|
|
) -> dict[str, dict[str, float]]:
|
|
frame = dataset.reset_index(drop=True)
|
|
groups = [group.index.to_numpy(dtype=np.int64) for _, group in frame.groupby("symbol", sort=False)]
|
|
if not groups:
|
|
return {}
|
|
rng = np.random.default_rng(seed)
|
|
values: dict[str, list[float]] = {"pinball_q20": [], "pinball_q80": [], "q20_hit_rate": [], "q80_coverage": []}
|
|
for _ in range(repetitions):
|
|
sampled: list[int] = []
|
|
for indices in groups:
|
|
if len(indices) == 0:
|
|
continue
|
|
need = len(indices)
|
|
local: list[int] = []
|
|
while len(local) < need:
|
|
start = int(rng.integers(0, len(indices)))
|
|
local.extend(indices[start : start + block_size].tolist())
|
|
sampled.extend(local[:need])
|
|
if not sampled:
|
|
continue
|
|
metrics = evaluate_predictions(
|
|
y_true[np.asarray(sampled)],
|
|
point[np.asarray(sampled)],
|
|
lower[np.asarray(sampled)],
|
|
upper[np.asarray(sampled)],
|
|
)
|
|
for name in values:
|
|
values[name].append(metrics[name])
|
|
return {
|
|
name: {
|
|
"lower_95": float(np.quantile(items, 0.025)),
|
|
"upper_95": float(np.quantile(items, 0.975)),
|
|
}
|
|
for name, items in values.items()
|
|
if items
|
|
}
|
|
|
|
|
|
def _json_safe(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
return {str(key): _json_safe(item) for key, item in value.items()}
|
|
if isinstance(value, (list, tuple)):
|
|
return [_json_safe(item) for item in value]
|
|
if isinstance(value, (np.integer, np.floating)):
|
|
return value.item()
|
|
return value
|
|
|
|
|
|
def run_training(
|
|
data_dir: Path,
|
|
artifact_dir: Path,
|
|
end_ms: int | None = None,
|
|
test_months: int = 10,
|
|
seed: int = 42,
|
|
n_estimators: int = 250,
|
|
n_jobs: int = -1,
|
|
volatility_rank_threshold: float = VOLATILITY_RANK_THRESHOLD,
|
|
) -> 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")
|
|
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"]))
|
|
test_start_ms = months_before(end_ms, test_months)
|
|
profile_indicators = {symbol: compute_indicators(frame) for symbol, frame in frames.items()}
|
|
raw_profiles = {
|
|
symbol: asset_profile(indicators, test_start_ms)
|
|
for symbol, indicators in profile_indicators.items()
|
|
}
|
|
profiles = {symbol: profile for symbol, profile in raw_profiles.items() if profile is not None}
|
|
if len(profiles) < 3:
|
|
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)
|
|
_progress(
|
|
f"training: {len(frames)} symbols, {len(profiles)} profiles; "
|
|
f"train/validation/test symbols={len(train_symbols)}/{len(validation_symbols)}/{len(test_symbols)}"
|
|
)
|
|
_progress(
|
|
f"training: ATR rank filter window={VOLATILITY_RANK_WINDOW}, "
|
|
f"threshold={volatility_rank_threshold:.2f}"
|
|
)
|
|
stage_results: list[dict[str, Any]] = []
|
|
best_overall: dict[str, Any] | None = None
|
|
best_dataset: pd.DataFrame | None = None
|
|
best_stage = 0
|
|
for stage in (1, 2, 3):
|
|
_progress(f"stage {stage}/3: building feature dataset")
|
|
dataset, feature_names, _ = build_feature_dataset(
|
|
frames,
|
|
stage,
|
|
test_start_ms,
|
|
end_ms,
|
|
asset_scaler,
|
|
train_symbols,
|
|
validation_symbols,
|
|
test_symbols,
|
|
validation_end_ms,
|
|
calibration_end_ms,
|
|
volatility_rank_threshold,
|
|
)
|
|
filter_stats = dataset.attrs.get("volatility_filter", {})
|
|
_progress(
|
|
f"stage {stage}/3: features={len(feature_names)}, rows="
|
|
f"train/validation/calibration/test={int((dataset['split'] == 'train').sum())}/"
|
|
f"{int((dataset['split'] == 'validation').sum())}/"
|
|
f"{int((dataset['split'] == 'calibration').sum())}/"
|
|
f"{int((dataset['split'] == 'test').sum())}; "
|
|
f"filter candidates={sum(filter_stats.get('candidate_counts', {}).values())}, "
|
|
f"selected={sum(filter_stats.get('selected_counts', {}).values())}"
|
|
)
|
|
best, candidate_results = _fit_stage(
|
|
dataset,
|
|
feature_names,
|
|
validation_symbols,
|
|
stage,
|
|
seed,
|
|
n_estimators,
|
|
n_jobs,
|
|
)
|
|
stage_record = {
|
|
"stage": stage,
|
|
"feature_count": len(feature_names),
|
|
"train_rows": int((dataset["split"] == "train").sum()),
|
|
"validation_rows": int((dataset["split"] == "validation").sum()),
|
|
"calibration_rows": int((dataset["split"] == "calibration").sum()),
|
|
"test_rows": int((dataset["split"] == "test").sum()),
|
|
"volatility_filter": filter_stats,
|
|
"best_model": best["spec"]["name"],
|
|
"best_validation": best["validation_metrics"],
|
|
"candidates": candidate_results,
|
|
}
|
|
stage_results.append(stage_record)
|
|
_progress(
|
|
f"stage {stage}/3: best={best['spec']['name']}, "
|
|
f"validation selection score={best['selection_score']:.6f}"
|
|
)
|
|
if best_overall is None or best["selection_score"] < best_overall["selection_score"]:
|
|
best_overall = best
|
|
best_dataset = dataset
|
|
best_stage = stage
|
|
if best_overall is None or best_dataset is None:
|
|
raise RuntimeError("no stage was trained")
|
|
_progress(f"training: selected stage {best_stage}; fitting final calibration/test report")
|
|
best_features = best_overall["feature_names"]
|
|
validation = best_dataset[
|
|
best_dataset["split"] == "validation"
|
|
]
|
|
calibration = best_dataset[best_dataset["split"] == "calibration"]
|
|
test = best_dataset[
|
|
best_dataset["split"] == "test"
|
|
]
|
|
if calibration.empty or test.empty:
|
|
raise ValueError("calibration and final test samples are required")
|
|
offsets = _calibration_offsets(best_overall, calibration)
|
|
test_x = test[best_features]
|
|
test_y = test["target"].to_numpy(dtype=float)
|
|
point, lower, upper = predict_quantiles(
|
|
best_overall["point_model"],
|
|
best_overall["lower_model"],
|
|
best_overall["upper_model"],
|
|
test_x,
|
|
best_overall["spec"]["kind"],
|
|
)
|
|
calibrated_lower, calibrated_upper = apply_calibration(lower, upper, offsets)
|
|
test_scale = test["target_scale"].to_numpy(dtype=float)
|
|
actual_test_y = test_y * test_scale
|
|
actual_point = point * test_scale
|
|
actual_calibrated_lower = calibrated_lower * test_scale
|
|
actual_calibrated_upper = calibrated_upper * test_scale
|
|
test_metrics = {
|
|
**evaluate_predictions(
|
|
test_y,
|
|
point,
|
|
calibrated_lower,
|
|
calibrated_upper,
|
|
target_scale=test_scale,
|
|
),
|
|
"raw_q80_coverage": float(np.mean(test_y <= upper)),
|
|
"raw_q20_hit_rate": float(np.mean(test_y >= lower)),
|
|
"calibrated_q80_coverage": float(np.mean(test_y <= calibrated_upper)),
|
|
"calibrated_q20_hit_rate": float(np.mean(test_y >= calibrated_lower)),
|
|
"raw_quantile_cross_rate": float(np.mean(lower > upper)),
|
|
}
|
|
test_metrics["stratified"] = _stratified_metrics(
|
|
test,
|
|
actual_test_y,
|
|
actual_point,
|
|
actual_calibrated_lower,
|
|
actual_calibrated_upper,
|
|
)
|
|
test_metrics["bootstrap_95_ci"] = _block_bootstrap_ci(
|
|
test,
|
|
actual_test_y,
|
|
actual_point,
|
|
actual_calibrated_lower,
|
|
actual_calibrated_upper,
|
|
seed,
|
|
)
|
|
symbol_prediction = pd.DataFrame(
|
|
{
|
|
"symbol": test["symbol"].to_numpy(),
|
|
"point": actual_point,
|
|
"q20": actual_calibrated_lower,
|
|
"q80": actual_calibrated_upper,
|
|
}
|
|
).groupby("symbol", sort=True).median()
|
|
test_metrics["prediction_dispersion_by_symbol"] = {
|
|
name: {
|
|
"min": float(symbol_prediction[name].min()),
|
|
"median": float(symbol_prediction[name].median()),
|
|
"max": float(symbol_prediction[name].max()),
|
|
"std": float(symbol_prediction[name].std(ddof=0)),
|
|
}
|
|
for name in ("point", "q20", "q80")
|
|
}
|
|
_progress(
|
|
f"training: final test rows={len(test)}, calibrated q80 coverage="
|
|
f"{test_metrics['calibrated_q80_coverage']:.4f}"
|
|
)
|
|
all_target = best_dataset["target"].to_numpy(dtype=float)
|
|
all_target_scale = best_dataset["target_scale"].to_numpy(dtype=float)
|
|
actual_target = all_target * all_target_scale
|
|
target_summary = {
|
|
"n": int(len(actual_target)),
|
|
"positive_fraction": float(np.mean(actual_target > 0)),
|
|
"negative_fraction": float(np.mean(actual_target < 0)),
|
|
"zero_fraction": float(np.mean(actual_target == 0)),
|
|
"quantiles": {
|
|
str(quantile): float(np.quantile(actual_target, quantile))
|
|
for quantile in (0.01, 0.10, 0.50, 0.90, 0.99)
|
|
},
|
|
}
|
|
symbol_files = {
|
|
item["symbol"]: item["path"]
|
|
for item in manifest.get("symbols", [])
|
|
if item["symbol"] in profiles
|
|
}
|
|
artifact_dir.mkdir(parents=True, exist_ok=True)
|
|
artifact = {
|
|
"version": 3,
|
|
"timeframe": "8h",
|
|
"window": WINDOW,
|
|
"horizon": HORIZON,
|
|
"target_kind": TARGET_KIND,
|
|
"target": "quantile(future_6_low, 0.20) / anchor_close - 1",
|
|
"future_low_quantile": FUTURE_LOW_QUANTILE,
|
|
"target_scale": "asset_atr_price_top20_median",
|
|
"lower_quantile": LOWER_QUANTILE,
|
|
"upper_quantile": TARGET_QUANTILE,
|
|
"volatility_rank_window": VOLATILITY_RANK_WINDOW,
|
|
"volatility_rank_threshold": volatility_rank_threshold,
|
|
"stage": best_stage,
|
|
"model_name": best_overall["spec"]["name"],
|
|
"model_kind": best_overall["spec"]["kind"],
|
|
"feature_names": best_features,
|
|
"asset_scaler": asset_scaler,
|
|
"asset_profiles": profiles,
|
|
"symbol_files": symbol_files,
|
|
"calibration": offsets,
|
|
"point_model": best_overall["point_model"],
|
|
"lower_model": best_overall["lower_model"],
|
|
"upper_model": best_overall["upper_model"],
|
|
"end_ms": end_ms,
|
|
"test_start_ms": test_start_ms,
|
|
"symbols": sorted(profiles),
|
|
}
|
|
joblib.dump(artifact, artifact_dir / "model.joblib")
|
|
report = {
|
|
"protocol": {
|
|
"train_anchor_before": test_start_ms,
|
|
"recent_window_start": test_start_ms,
|
|
"recent_window_end": end_ms,
|
|
"validation_end": validation_end_ms,
|
|
"calibration_end": calibration_end_ms,
|
|
"final_test_start": calibration_end_ms,
|
|
"test_months": test_months,
|
|
"validation_symbols": validation_symbols,
|
|
"test_symbols": test_symbols,
|
|
"training_symbols_with_history": train_symbols,
|
|
"seed": seed,
|
|
"future_bars": HORIZON,
|
|
"feature_window_bars": WINDOW,
|
|
"target": {
|
|
"kind": TARGET_KIND,
|
|
"future_low_quantile": FUTURE_LOW_QUANTILE,
|
|
"scale": "asset_atr_price_top20_median",
|
|
"summary": target_summary,
|
|
},
|
|
"volatility_filter": {
|
|
"atr_rank_window": VOLATILITY_RANK_WINDOW,
|
|
"min_rank": volatility_rank_threshold,
|
|
},
|
|
},
|
|
"selected": {
|
|
"stage": best_stage,
|
|
"model": best_overall["spec"]["name"],
|
|
"model_kind": best_overall["spec"]["kind"],
|
|
"feature_count": len(best_features),
|
|
"validation": best_overall["validation_metrics"],
|
|
"calibration_rows": len(calibration),
|
|
"calibration": offsets,
|
|
},
|
|
"stages": stage_results,
|
|
"test": test_metrics,
|
|
"feature_names": best_features,
|
|
"artifact": str((artifact_dir / "model.joblib").resolve()),
|
|
}
|
|
report = _json_safe(report)
|
|
(artifact_dir / "report.json").write_text(json.dumps(report, indent=2, ensure_ascii=False))
|
|
(artifact_dir / "report.md").write_text(render_report(report))
|
|
_progress(f"training: artifacts written to {artifact_dir}")
|
|
return report
|
|
|
|
|
|
def build_prediction_features(
|
|
frame: pd.DataFrame,
|
|
stage: int,
|
|
anchor_index: int,
|
|
asset_scaler: dict[str, dict[str, float]],
|
|
profile: dict[str, float] | None = None,
|
|
) -> pd.DataFrame:
|
|
indicators = compute_indicators(frame).reset_index(drop=True)
|
|
if profile is None:
|
|
profile = asset_profile(indicators, int(indicators.iloc[anchor_index]["timestamp"]) + 1)
|
|
if profile is None:
|
|
raise ValueError("prediction history is too short to calculate asset profile")
|
|
asset = scale_asset_profile(profile, asset_scaler)
|
|
if anchor_index < WINDOW - 1:
|
|
raise ValueError(f"at least {WINDOW} closed 8H bars are required before anchor")
|
|
window = indicators.iloc[anchor_index - WINDOW + 1 : anchor_index + 1]
|
|
values = _window_features(window, stage, asset)
|
|
return pd.DataFrame([values])
|
|
|
|
|
|
def predict_from_csv(
|
|
artifact_path: Path,
|
|
csv_path: Path,
|
|
at_ms: int | None = None,
|
|
symbol: str | None = None,
|
|
) -> dict[str, Any]:
|
|
artifact = joblib.load(artifact_path)
|
|
if (
|
|
artifact.get("version") != 3
|
|
or artifact.get("target_kind") != TARGET_KIND
|
|
or artifact.get("target_scale") != "asset_atr_price_top20_median"
|
|
):
|
|
raise ValueError(
|
|
"artifact uses a legacy target; retrain with the signed robust future-low target"
|
|
)
|
|
frame = pd.read_csv(csv_path)
|
|
for column in CSV_COLUMNS:
|
|
frame[column] = pd.to_numeric(frame[column], errors="coerce")
|
|
frame = frame.sort_values("timestamp").drop_duplicates("timestamp", keep="last").reset_index(drop=True)
|
|
frame["quote"] = np.where(frame["quote"] > 0, frame["quote"], frame["volume"] * frame["close"])
|
|
if at_ms is None:
|
|
anchor_index = len(frame) - 1
|
|
else:
|
|
if int(at_ms) < int(artifact.get("test_start_ms", 0)):
|
|
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")
|
|
anchor_index = int(matches[-1])
|
|
if symbol is None:
|
|
for candidate, path in artifact.get("symbol_files", {}).items():
|
|
if Path(path).name == csv_path.name:
|
|
symbol = candidate
|
|
break
|
|
profile = artifact.get("asset_profiles", {}).get(symbol) if symbol else None
|
|
if profile is None:
|
|
raise ValueError("symbol must match a symbol with a stored training asset profile")
|
|
indicators = compute_indicators(frame)
|
|
rank_value = float(indicators.iloc[anchor_index]["atr_rank_200"])
|
|
threshold = float(artifact.get("volatility_rank_threshold", VOLATILITY_RANK_THRESHOLD))
|
|
eligible = np.isfinite(rank_value) and rank_value >= threshold
|
|
target_scale = max(float(profile["asset_atr_price_top20_median"]), 1e-9)
|
|
response = {
|
|
"timestamp": int(frame.iloc[anchor_index]["timestamp"]),
|
|
"close": float(frame.iloc[anchor_index]["close"]),
|
|
"model": artifact["model_name"],
|
|
"feature_stage": artifact["stage"],
|
|
"symbol": symbol,
|
|
"volatility_rank_window": VOLATILITY_RANK_WINDOW,
|
|
"volatility_rank_200": rank_value if np.isfinite(rank_value) else None,
|
|
"volatility_rank_threshold": threshold,
|
|
"future_low_quantile": float(artifact.get("future_low_quantile", FUTURE_LOW_QUANTILE)),
|
|
"target_scale": target_scale,
|
|
"eligible_for_model": bool(eligible),
|
|
}
|
|
if not eligible:
|
|
response["skip_reason"] = (
|
|
"atr_rank_200_unavailable"
|
|
if not np.isfinite(rank_value)
|
|
else "atr_rank_200_below_threshold"
|
|
)
|
|
response.update(
|
|
{
|
|
"drawdown_point": None,
|
|
"drawdown_hit80_q20": None,
|
|
"drawdown_upper80_q80": None,
|
|
"price_level_hit80": None,
|
|
"price_level_upper80": None,
|
|
"future_low_return_point": None,
|
|
"future_low_return_q20": None,
|
|
"future_low_return_q80": None,
|
|
"future_low_reference_return_point": None,
|
|
"future_low_reference_return_q20": None,
|
|
"future_low_reference_return_q80": None,
|
|
"future_low_price_q20": None,
|
|
"future_low_price_q80": None,
|
|
"future_low_reference_price_q20": None,
|
|
"future_low_reference_price_q80": None,
|
|
}
|
|
)
|
|
return response
|
|
features = build_prediction_features(
|
|
frame,
|
|
artifact["stage"],
|
|
anchor_index,
|
|
artifact["asset_scaler"],
|
|
profile,
|
|
)
|
|
point, lower, upper = predict_quantiles(
|
|
artifact["point_model"],
|
|
artifact["lower_model"],
|
|
artifact["upper_model"],
|
|
features,
|
|
artifact["model_kind"],
|
|
)
|
|
lower, upper = apply_calibration(lower, upper, artifact.get("calibration", {}))
|
|
signed_point = float(point[0]) * target_scale
|
|
signed_q20 = float(lower[0]) * target_scale
|
|
signed_q80 = float(upper[0]) * target_scale
|
|
close = response["close"]
|
|
response.update(
|
|
{
|
|
"future_low_return_point": signed_point,
|
|
"future_low_return_q20": signed_q20,
|
|
"future_low_return_q80": signed_q80,
|
|
"future_low_reference_return_point": signed_point,
|
|
"future_low_reference_return_q20": signed_q20,
|
|
"future_low_reference_return_q80": signed_q80,
|
|
"future_low_price_q20": close * (1 + signed_q20),
|
|
"future_low_price_q80": close * (1 + signed_q80),
|
|
"future_low_reference_price_q20": close * (1 + signed_q20),
|
|
"future_low_reference_price_q80": close * (1 + signed_q80),
|
|
"drawdown_point": -signed_point,
|
|
"drawdown_hit80_q20": -signed_q80,
|
|
"drawdown_upper80_q80": -signed_q20,
|
|
"price_level_hit80": close * (1 + signed_q80),
|
|
"price_level_upper80": close * (1 + signed_q20),
|
|
}
|
|
)
|
|
return response
|
|
|
|
|
|
def render_report(report: dict[str, Any]) -> str:
|
|
selected = report["selected"]
|
|
test = report["test"]
|
|
volatility_filter = report["protocol"]["volatility_filter"]
|
|
lines = [
|
|
"# cbpred report",
|
|
"",
|
|
f"- selected stage: {selected['stage']}",
|
|
f"- selected model: `{selected['model']}`",
|
|
f"- feature count: {selected['feature_count']}",
|
|
"- target: signed 20th-percentile future-low return, normalized by each asset's training ATR/price scale",
|
|
f"- target positive/negative fraction: {report['protocol']['target']['summary']['positive_fraction']:.4f}/"
|
|
f"{report['protocol']['target']['summary']['negative_fraction']:.4f}",
|
|
f"- volatility filter: ATR rank over {volatility_filter['atr_rank_window']} bars >= "
|
|
f"{volatility_filter['min_rank']:.2f}",
|
|
f"- validation q80 coverage: {selected['validation']['q80_coverage']:.4f}",
|
|
f"- test calibrated q20 pinball: {test['pinball_q20']:.6f}",
|
|
f"- test calibrated q80 pinball: {test['pinball_q80']:.6f}",
|
|
f"- test calibrated q80 coverage: {test['calibrated_q80_coverage']:.4f}",
|
|
f"- test calibrated q20 lower-bound survival rate: {test['calibrated_q20_hit_rate']:.4f}",
|
|
f"- test point MAE: {test['mae_point']:.6f}",
|
|
f"- test point RMSE: {test['rmse_point']:.6f}",
|
|
f"- test q80 pinball loss: {test['pinball_q80']:.6f}",
|
|
f"- test per-symbol median point return range: "
|
|
f"{test['prediction_dispersion_by_symbol']['point']['min']:.6f} - "
|
|
f"{test['prediction_dispersion_by_symbol']['point']['max']:.6f}",
|
|
"",
|
|
"The target is quantile(future_6_low, 0.20) / anchor_close - 1. Negative values are drawdowns; "
|
|
"positive values mean the robust low reference stayed above the anchor close. "
|
|
"q20 is the lower future-low return quantile and therefore the conservative lower price level; "
|
|
"q80 is the shallower, more optimistic level. Coverage is calibrated on the held-out test protocol.",
|
|
"",
|
|
"## Three feature iterations",
|
|
"",
|
|
]
|
|
for item in report["stages"]:
|
|
best = item["best_validation"]
|
|
lines.append(
|
|
f"- stage {item['stage']}: {item['feature_count']} features, "
|
|
f"best `{item['best_model']}`, validation q20/q80 pinball="
|
|
f"{best['pinball_q20']:.6f}/{best['pinball_q80']:.6f}, "
|
|
f"q80 coverage={best['q80_coverage']:.4f}"
|
|
)
|
|
ci = test.get("bootstrap_95_ci", {})
|
|
if ci:
|
|
lines.extend(["", "## Final test block-bootstrap 95% intervals", ""])
|
|
for name, interval in ci.items():
|
|
label = (
|
|
"q20 lower-bound survival rate"
|
|
if name == "q20_hit_rate"
|
|
else name
|
|
)
|
|
lines.append(f"- {label}: {interval['lower_95']:.6f} - {interval['upper_95']:.6f}")
|
|
lines.extend(["", "## Feature names", "", "```text", *report["feature_names"], "```", ""])
|
|
return "\n".join(lines)
|