start experiment 38 (exp/38-q06-kelly-sizing-score-magnitude-fractio)
This commit is contained in:
zhaoli
@@ -0,0 +1,25 @@
|
||||
from .config import (
|
||||
LakeConfig,
|
||||
BAR_FIELD_MAP,
|
||||
FREQ_TO_TIMEFRAME,
|
||||
UNKNOWN_FIELD_NAMES,
|
||||
timeframe_for_freq,
|
||||
resolve_lake_root,
|
||||
)
|
||||
from .providers import (
|
||||
LakeCalendarProvider,
|
||||
LakeInstrumentProvider,
|
||||
LakeFeatureProvider,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"LakeConfig",
|
||||
"BAR_FIELD_MAP",
|
||||
"FREQ_TO_TIMEFRAME",
|
||||
"UNKNOWN_FIELD_NAMES",
|
||||
"timeframe_for_freq",
|
||||
"resolve_lake_root",
|
||||
"LakeCalendarProvider",
|
||||
"LakeInstrumentProvider",
|
||||
"LakeFeatureProvider",
|
||||
]
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,202 @@
|
||||
"""TradeAC lake configuration helpers.
|
||||
|
||||
The lake is a hive-partitioned parquet store (see ``tac-engine/skills/tradeac-lake``):
|
||||
|
||||
$TAC_LAKE_DIR/
|
||||
├── market=US/
|
||||
│ └── timeframe=1d/
|
||||
│ └── symbol=AAPL.parquet # OHLCV bars: t, date, o, h, l, c, v, n, vw
|
||||
├── features/ # indicators, wide format, family tier
|
||||
│ └── market=US/
|
||||
│ └── timeframe=1d/
|
||||
│ ├── family=ta/symbol=AAPL.parquet # t, sma_5, sma_20, rsi_14, ...
|
||||
│ └── family=sp/symbol=AAPL.parquet # t, sp_ou_*, sp_hmm_*, ...
|
||||
├── calendar.parquet # trading days per market
|
||||
├── coverage.parquet # per (market,timeframe,symbol) loaded windows
|
||||
└── symbols.parquet # asset master
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
import pandas as pd
|
||||
|
||||
#: qlib freq string (Freq.__str__) -> lake timeframe partition name
|
||||
FREQ_TO_TIMEFRAME: Dict[str, str] = {
|
||||
"day": "1d",
|
||||
"1d": "1d",
|
||||
"min": "1m",
|
||||
"1min": "1m",
|
||||
"5min": "5m",
|
||||
"10min": "10m",
|
||||
"15min": "15m",
|
||||
"30min": "30m",
|
||||
"hour": "1h",
|
||||
"1hour": "1h",
|
||||
"2hour": "2h",
|
||||
"4hour": "4h",
|
||||
"week": "1w",
|
||||
"1week": "1w",
|
||||
"month": "1M",
|
||||
"1month": "1M",
|
||||
}
|
||||
|
||||
#: bar-field map: qlib field name (without the leading ``$``) -> lake bar column
|
||||
BAR_FIELD_MAP: Dict[str, str] = {
|
||||
"open": "o",
|
||||
"high": "h",
|
||||
"low": "l",
|
||||
"close": "c",
|
||||
"volume": "v",
|
||||
"vwap": "vw",
|
||||
"avg_amount": "vw", # amount / volume
|
||||
}
|
||||
|
||||
#: fields that qlib core/backtest queries but the lake does not store -> all-NaN
|
||||
UNKNOWN_FIELD_NAMES = ("factor", "change", "trade_unit", "suspend_flag")
|
||||
|
||||
#: columns in the parquet files that are not features
|
||||
NON_FEATURE_COLUMNS = ("t", "date", "market", "timeframe", "symbol")
|
||||
|
||||
|
||||
def timeframe_for_freq(freq: str) -> str:
|
||||
"""Map a qlib frequency (e.g. ``day``, ``1min``) to a lake timeframe (e.g. ``1d``)."""
|
||||
f = str(freq).lower()
|
||||
if f not in FREQ_TO_TIMEFRAME:
|
||||
raise ValueError(
|
||||
f"unsupported qlib freq {freq!r}; supported freqs: {sorted(set(FREQ_TO_TIMEFRAME))}"
|
||||
)
|
||||
return FREQ_TO_TIMEFRAME[f]
|
||||
|
||||
|
||||
def resolve_lake_root(lake_root: Optional[str] = None) -> Path:
|
||||
"""Resolve the lake root: explicit arg > ``TAC_LAKE_DIR`` (no fallback).
|
||||
|
||||
``TAC_LAKE_DIR`` is **mandatory** — there is deliberately no default
|
||||
A missing/empty value raises so a
|
||||
misconfigured environment never silently points at a wrong directory.
|
||||
"""
|
||||
if lake_root is None:
|
||||
lake_root = os.environ.get("TAC_LAKE_DIR")
|
||||
if not lake_root:
|
||||
raise RuntimeError(
|
||||
"TAC_LAKE_DIR is not set. Point it at the TradeAC lake root, e.g. "
|
||||
"export TAC_LAKE_DIR=/home/data/lake (docker) or set an absolute "
|
||||
"path in your local .env."
|
||||
)
|
||||
return Path(str(lake_root)).expanduser().resolve()
|
||||
|
||||
|
||||
class LakeConfig:
|
||||
"""Path helpers + cached readers for a (lake_root, market) combination."""
|
||||
|
||||
def __init__(self, lake_root: Optional[str] = None, market: str = "US"):
|
||||
self.lake_root: Path = resolve_lake_root(lake_root)
|
||||
self.market: str = (market or "US").upper()
|
||||
|
||||
# ---- paths --------------------------------------------------------------
|
||||
def bar_dir(self, timeframe: str) -> Path:
|
||||
return self.lake_root / f"market={self.market}" / f"timeframe={timeframe}"
|
||||
|
||||
def bar_path(self, timeframe: str, symbol: str) -> Path:
|
||||
return self.bar_dir(timeframe) / f"symbol={str(symbol).upper()}.parquet"
|
||||
|
||||
def features_dir(self, timeframe: str) -> Path:
|
||||
return self.lake_root / "features" / f"market={self.market}" / f"timeframe={timeframe}"
|
||||
|
||||
def features_path(self, timeframe: str, symbol: str) -> Path:
|
||||
# Legacy flat path (no family tier). Prefer `load_features` which
|
||||
# resolves the family=ta|sp partition layout.
|
||||
return self.features_dir(timeframe) / f"symbol={str(symbol).upper()}.parquet"
|
||||
|
||||
def load_features(self, timeframe: str, symbol: str) -> pd.DataFrame:
|
||||
"""All feature columns for a symbol, merging the `family=ta` and
|
||||
`family=sp` partitions by timestamp. Returns an empty frame when no
|
||||
feature files exist (legacy flat layout falls back transparently)."""
|
||||
sym = str(symbol).upper()
|
||||
frames = []
|
||||
for family in ("ta", "sp"):
|
||||
p = self.features_dir(timeframe) / f"family={family}" / f"symbol={sym}.parquet"
|
||||
if p.exists():
|
||||
frames.append(pd.read_parquet(p))
|
||||
if not frames:
|
||||
flat = self.features_dir(timeframe) / f"symbol={sym}.parquet"
|
||||
if flat.exists():
|
||||
return pd.read_parquet(flat)
|
||||
return pd.DataFrame()
|
||||
if len(frames) == 1:
|
||||
return frames[0]
|
||||
merged = frames[0]
|
||||
for extra in frames[1:]:
|
||||
merged = merged.merge(extra, on="t", how="outer", suffixes=("", "_dup"))
|
||||
for c in [c for c in merged.columns if c.endswith("_dup")]:
|
||||
merged = merged.drop(columns=c)
|
||||
return merged
|
||||
|
||||
def calendar_path(self) -> Path:
|
||||
return self.lake_root / "calendar.parquet"
|
||||
|
||||
def symbols_path(self) -> Path:
|
||||
return self.lake_root / "symbols.parquet"
|
||||
|
||||
def coverage_path(self) -> Path:
|
||||
return self.lake_root / "coverage.parquet"
|
||||
|
||||
# ---- metadata readers ----------------------------------------------------
|
||||
def load_symbols(self) -> List[str]:
|
||||
"""All symbols known to the lake (from ``symbols.parquet``)."""
|
||||
p = self.symbols_path()
|
||||
if not p.exists():
|
||||
return []
|
||||
df = pd.read_parquet(p)
|
||||
if "symbol" not in df.columns:
|
||||
return []
|
||||
return sorted(df["symbol"].astype(str).str.upper().tolist())
|
||||
|
||||
def symbol_spans(self, symbol: str, timeframe: str) -> List[tuple]:
|
||||
"""Listing span(s) ``[(start_iso, end_iso)]`` for a symbol from coverage.parquet."""
|
||||
p = self.coverage_path()
|
||||
if p.exists():
|
||||
try:
|
||||
df = pd.read_parquet(p)
|
||||
except Exception: # pragma: no cover - defensive
|
||||
df = pd.DataFrame()
|
||||
if len(df):
|
||||
df = df[
|
||||
(df.get("market") == self.market)
|
||||
& (df.get("timeframe") == timeframe)
|
||||
& (df.get("symbol") == str(symbol).upper())
|
||||
]
|
||||
if len(df):
|
||||
row = df.iloc[0]
|
||||
first = pd.Timestamp(row["first_t"]).date()
|
||||
last = pd.Timestamp(row["last_t"]).date()
|
||||
return [(first.isoformat(), last.isoformat())]
|
||||
# fallback: derive from the bar file itself
|
||||
p = self.bar_path(timeframe, symbol)
|
||||
if p.exists():
|
||||
import pyarrow.parquet as pq
|
||||
|
||||
tbl = pq.read_table(p, columns=["t"])
|
||||
first = pd.Timestamp(tbl.column("t")[0].as_py()).date()
|
||||
last = pd.Timestamp(tbl.column("t")[-1].as_py()).date()
|
||||
return [(first.isoformat(), last.isoformat())]
|
||||
return [("1970-01-01", "2099-12-31")]
|
||||
|
||||
def load_calendar_dates(self) -> List[pd.Timestamp]:
|
||||
"""Trading days (midnight timestamps) for the market, from ``calendar.parquet``."""
|
||||
p = self.calendar_path()
|
||||
if p.exists():
|
||||
df = pd.read_parquet(p)
|
||||
if "date" in df.columns:
|
||||
if "market" in df.columns:
|
||||
df = df[df["market"] == self.market]
|
||||
dates = pd.to_datetime(df["date"]).dt.normalize().sort_values().unique()
|
||||
return [pd.Timestamp(x) for x in dates]
|
||||
return []
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover
|
||||
return f"LakeConfig(lake_root={self.lake_root}, market={self.market})"
|
||||
@@ -0,0 +1,230 @@
|
||||
"""qlib data providers backed by the TradeAC parquet lake.
|
||||
|
||||
These providers plug into the standard qlib mechanism: ``qlib.init(calendar_provider=...,
|
||||
instrument_provider=..., feature_provider=...)`` instantiates them and binds them to the
|
||||
``Cal`` / ``Inst`` / ``FeatureD`` wrappers (see ``qlib.data.data.register_all_wrappers``).
|
||||
The rest of qlib (``LocalDatasetProvider`` expression engine, backtest ``Exchange``) keeps
|
||||
working unchanged because the interface contract is identical to the file-based providers:
|
||||
|
||||
- ``feature()`` returns a ``pd.Series`` indexed by the **calendar position** range
|
||||
``[start_index, end_index]`` (matching ``FileFeatureStorage.__getitem__`` semantics).
|
||||
- ``list_instruments()`` returns ``{symbol: [(start, end), ...]}``.
|
||||
- ``load_calendar()`` returns a list of ``pd.Timestamp`` trading days.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import bisect
|
||||
from typing import Dict, List, Optional, Union
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from qlib.data.data import CalendarProvider, FeatureProvider, InstrumentProvider
|
||||
from qlib.log import get_module_logger
|
||||
|
||||
from .config import (
|
||||
BAR_FIELD_MAP,
|
||||
LakeConfig,
|
||||
UNKNOWN_FIELD_NAMES,
|
||||
timeframe_for_freq,
|
||||
)
|
||||
|
||||
logger = get_module_logger("tac_qlib.data.providers")
|
||||
|
||||
|
||||
def _day_freq(freq: str) -> bool:
|
||||
return str(freq).lower() in ("day", "1d")
|
||||
|
||||
|
||||
def _calendar_keys(cal: List[pd.Timestamp], freq: str) -> pd.Index:
|
||||
"""Convert calendar timestamps into the same key space as the lake parquet."""
|
||||
if _day_freq(freq):
|
||||
return pd.Index([pd.Timestamp(x).date() for x in cal])
|
||||
return pd.Index([pd.Timestamp(x) for x in cal])
|
||||
|
||||
|
||||
class LakeCalendarProvider(CalendarProvider):
|
||||
"""Trading calendar read from ``<lake>/calendar.parquet`` (fallback: derived from bars)."""
|
||||
|
||||
def __init__(self, lake_root: Optional[str] = None, market: str = "US"):
|
||||
super().__init__()
|
||||
self.cfg = LakeConfig(lake_root, market)
|
||||
|
||||
def load_calendar(self, freq, future):
|
||||
timeframe = timeframe_for_freq(freq)
|
||||
if not _day_freq(freq):
|
||||
raise NotImplementedError(
|
||||
f"freq={freq!r} (timeframe={timeframe}) is not supported yet: the lake calendar "
|
||||
f"only covers daily sessions; add a minute-level calendar to `calendar.parquet`"
|
||||
)
|
||||
|
||||
dates = self.cfg.load_calendar_dates()
|
||||
if not dates:
|
||||
# Fallback: derive the trading-day set from the persisted bar files.
|
||||
bar_dir = self.cfg.bar_dir(timeframe)
|
||||
if bar_dir.exists():
|
||||
import pyarrow.parquet as pq
|
||||
|
||||
cal: Dict[pd.Timestamp, None] = {}
|
||||
for p in sorted(bar_dir.glob("symbol=*.parquet")):
|
||||
tbl = pq.read_table(p, columns=["t"])
|
||||
for v in tbl.column("t"):
|
||||
cal[pd.Timestamp(v.as_py()).normalize()] = None
|
||||
dates = sorted(cal.keys())
|
||||
if not dates:
|
||||
return []
|
||||
|
||||
if future:
|
||||
# append the next calendar day so that "today" is a valid trade date
|
||||
last = dates[-1]
|
||||
dates = dates + [pd.Timestamp(last) + pd.Timedelta(days=1)]
|
||||
return dates
|
||||
|
||||
|
||||
class LakeInstrumentProvider(InstrumentProvider):
|
||||
"""Instruments from ``<lake>/symbols.parquet`` with listing spans from ``coverage.parquet``."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
lake_root: Optional[str] = None,
|
||||
market: str = "US",
|
||||
markets: Optional[Dict[str, list]] = None,
|
||||
):
|
||||
super().__init__()
|
||||
self.cfg = LakeConfig(lake_root, market)
|
||||
#: optional named pools, e.g. ``{"sp500": ["AAPL", "MSFT"], "etf": ["SPY"]}``.
|
||||
#: ``all`` / any unregistered name resolves to every symbol in the lake.
|
||||
self.markets: Dict[str, list] = markets or {}
|
||||
|
||||
def _resolve_symbols(self, market: Union[str, list]) -> List[str]:
|
||||
if isinstance(market, (list, tuple, pd.Index, np.ndarray)):
|
||||
return [str(s).upper() for s in market]
|
||||
if isinstance(market, str) and "," in market:
|
||||
return [s.strip().upper() for s in market.split(",") if s.strip()]
|
||||
if market in self.markets:
|
||||
return [str(s).upper() for s in self.markets[market]]
|
||||
return self.cfg.load_symbols()
|
||||
|
||||
def list_instruments(self, instruments, start_time=None, end_time=None, freq="day", as_list=False):
|
||||
market = instruments["market"]
|
||||
timeframe = timeframe_for_freq(freq)
|
||||
|
||||
symbols = self._resolve_symbols(market)
|
||||
if not symbols:
|
||||
if as_list:
|
||||
return []
|
||||
return {}
|
||||
|
||||
# clip listing spans to the queried window (mirror of LocalInstrumentProvider)
|
||||
from qlib.data.data import Cal # pylint: disable=C0415
|
||||
|
||||
cal = Cal.calendar(freq=freq)
|
||||
start_time = pd.Timestamp(start_time or cal[0])
|
||||
end_time = pd.Timestamp(end_time or cal[-1])
|
||||
|
||||
out: Dict[str, list] = {}
|
||||
for symbol in symbols:
|
||||
spans = []
|
||||
for begin, end in self.cfg.symbol_spans(symbol, timeframe):
|
||||
lo = max(start_time, pd.Timestamp(begin))
|
||||
hi = min(end_time, pd.Timestamp(end))
|
||||
if lo <= hi:
|
||||
spans.append((lo, hi))
|
||||
if spans:
|
||||
out[symbol] = spans
|
||||
|
||||
filter_pipe = instruments.get("filter_pipe") or []
|
||||
for filter_config in filter_pipe:
|
||||
from qlib.data import filter as F # pylint: disable=C0415
|
||||
|
||||
filter_t = getattr(F, filter_config["filter_type"]).from_config(filter_config)
|
||||
out = filter_t(out, start_time, end_time, freq)
|
||||
|
||||
if as_list:
|
||||
return list(out)
|
||||
return out
|
||||
|
||||
|
||||
class LakeFeatureProvider(FeatureProvider):
|
||||
"""Feature data from the lake parquet (OHLCV bars + pre-computed ta-lib features).
|
||||
|
||||
Field routing:
|
||||
- ``$open/$high/$low/$close/$volume/$vwap`` -> bar parquet columns
|
||||
- ``$amount`` (= v*vw), ``$avg_amount`` (= vw) -> derived from bar parquet
|
||||
- ``$factor/$change/...`` -> all-NaN (not stored)
|
||||
- anything else -> a ta-lib column in the features parquet
|
||||
"""
|
||||
|
||||
def __init__(self, lake_root: Optional[str] = None, market: str = "US"):
|
||||
super().__init__()
|
||||
self.cfg = LakeConfig(lake_root, market)
|
||||
self._bar_cache: Dict[tuple, pd.DataFrame] = {}
|
||||
self._feature_cache: Dict[tuple, pd.DataFrame] = {}
|
||||
|
||||
# ------------------------------------------------------------------ caches
|
||||
def _load_bar_df(self, instrument: str, timeframe: str) -> pd.DataFrame:
|
||||
key = (instrument, timeframe)
|
||||
if key not in self._bar_cache:
|
||||
p = self.cfg.bar_path(timeframe, instrument)
|
||||
self._bar_cache[key] = pd.read_parquet(p) if p.exists() else pd.DataFrame()
|
||||
return self._bar_cache[key]
|
||||
|
||||
def _load_feature_df(self, instrument: str, timeframe: str) -> pd.DataFrame:
|
||||
key = (instrument, timeframe)
|
||||
if key not in self._feature_cache:
|
||||
self._feature_cache[key] = self.cfg.load_features(timeframe, instrument)
|
||||
return self._feature_cache[key]
|
||||
|
||||
@staticmethod
|
||||
def _keys(df: pd.DataFrame, freq: str) -> pd.Index:
|
||||
ts = pd.to_datetime(df["t"])
|
||||
return ts.dt.date if _day_freq(freq) else ts
|
||||
|
||||
# ------------------------------------------------------------------ fields
|
||||
def _extract(self, instrument: str, field: str, timeframe: str, freq: str) -> Optional[pd.Series]:
|
||||
"""Return the field as a Series keyed by date/timestamp (None if not present in the lake)."""
|
||||
bar = self._load_bar_df(instrument, timeframe)
|
||||
|
||||
if field in BAR_FIELD_MAP:
|
||||
col = BAR_FIELD_MAP[field]
|
||||
if col in bar.columns:
|
||||
return bar[col].astype(float).set_axis(self._keys(bar, freq))
|
||||
return None
|
||||
if field == "amount":
|
||||
if "v" in bar.columns and "vw" in bar.columns:
|
||||
return (bar["v"] * bar["vw"]).astype(float).set_axis(self._keys(bar, freq))
|
||||
return None
|
||||
if field in UNKNOWN_FIELD_NAMES:
|
||||
return None
|
||||
|
||||
feat = self._load_feature_df(instrument, timeframe)
|
||||
if field in feat.columns:
|
||||
return feat[field].astype(float).set_axis(self._keys(feat, freq))
|
||||
return None
|
||||
|
||||
# ------------------------------------------------------------------ api
|
||||
def _get_calendar(self, freq: str) -> List[pd.Timestamp]:
|
||||
from qlib.data.data import Cal # pylint: disable=C0415
|
||||
|
||||
cal = Cal.calendar(freq=freq)
|
||||
return list(cal)
|
||||
|
||||
def feature(self, instrument, field, start_index, end_index, freq):
|
||||
field = str(field)[1:]
|
||||
timeframe = timeframe_for_freq(freq)
|
||||
|
||||
cal = self._get_calendar(freq)
|
||||
n = len(cal)
|
||||
lo = max(0, int(start_index))
|
||||
hi = min(n - 1, int(end_index))
|
||||
if lo > hi:
|
||||
return pd.Series(dtype=np.float32)
|
||||
|
||||
keys = _calendar_keys(cal[lo : hi + 1], freq)
|
||||
ser = self._extract(str(instrument).upper(), field, timeframe, freq)
|
||||
if ser is None:
|
||||
vals = np.full(len(keys), np.nan, dtype=np.float64)
|
||||
else:
|
||||
vals = ser.reindex(keys).to_numpy(dtype=np.float64)
|
||||
return pd.Series(vals, index=pd.RangeIndex(lo, hi + 1))
|
||||
Reference in New Issue
Block a user