From 8d908477e27e7cbbd5fb338cc7c7e82a42d0d1c8 Mon Sep 17 00:00:00 2001 From: Yuxuan Yan Date: Tue, 16 Jun 2026 15:55:30 +0800 Subject: [PATCH] Add daily derived data pipeline --- README.md | 78 ++++++ cli.py | 3 + docs/minute_bar_data.md | 11 +- pipeline/alpha/cli.py | 2 +- pipeline/alpha/compute.py | 16 +- pipeline/common/schema.py | 7 + pipeline/derived/__init__.py | 2 + pipeline/derived/base.py | 38 +++ pipeline/derived/cli.py | 145 ++++++++++ pipeline/derived/compute.py | 115 ++++++++ pipeline/derived/library/__init__.py | 4 + .../derived/library/minute_daily_summary.py | 88 ++++++ pipeline/derived/registry.py | 80 ++++++ pipeline/features/base.py | 36 +-- pipeline/features/compute.py | 72 ++--- .../features/library/minute_daily_summary.py | 78 +----- pipeline/features/registry.py | 83 ++---- tests/test_derived.py | 261 ++++++++++++++++++ tests/test_features.py | 9 + 19 files changed, 897 insertions(+), 231 deletions(-) create mode 100644 pipeline/derived/__init__.py create mode 100644 pipeline/derived/base.py create mode 100644 pipeline/derived/cli.py create mode 100644 pipeline/derived/compute.py create mode 100644 pipeline/derived/library/__init__.py create mode 100644 pipeline/derived/library/minute_daily_summary.py create mode 100644 pipeline/derived/registry.py create mode 100644 tests/test_derived.py diff --git a/README.md b/README.md index 298593f..b4826dc 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,49 @@ partitions already written. Pass the **dataset directory** (`{output_dir}/{unive as `--data-path` to later phases — `pd.read_parquet` reads the whole partitioned set. Symbols use the internal `sh600000` / `sz000001` form (exchange prefix + code). +### `derived` — daily custom/derived data + +Derived data is daily-only v1 research data keyed by `symbol_id,date`, with one +or more numeric value columns. It can come from user CSV/parquet files or Python +plugins, and is written as a single parquet file at `derived/{name}.pq`. + +The validator normalizes `date` to the trading day, requires unique +`symbol_id,date` keys, rejects duplicate columns, and rejects non-numeric value +columns. Alpha computation consumes derived data through the existing +`--feature-path` flag. + +```bash +# Validate a user file without writing output. +uv run python cli.py derived validate --input-path vendor_factor.csv + +# Ingest CSV/parquet into the canonical derived/ layout. +uv run python cli.py derived ingest \ + --input-path vendor_factor.csv \ + --derived-name vendor_factor + +# List built-in and external derived-data plugin types. +uv run python cli.py derived list +uv run python cli.py derived list --derived-module path/to/my_derived.py + +# Compute a derived file from daily and/or minute inputs. +uv run python cli.py derived compute \ + --minute-path data/minute_bars/sh600000 \ + --daily-path data/daily_bars/sh600000 \ + --derived-type minute_daily_summary \ + --derived-name minute_summary + +# Join derived columns into a feature-aware alpha. +uv run python cli.py alpha compute \ + --data-path data/daily_bars/sh600000 \ + --feature-path derived/minute_summary.pq \ + --alpha-type my_feature_aware_alpha \ + --alpha-name my_run +``` + +For compatibility, `feature list` and `feature compute` remain available and +delegate to the same derived-data registry. Existing `features/*.pq` files are +still valid `--feature-path` inputs when they satisfy the daily numeric contract. + ### `alpha list` — show registered alpha types ```bash @@ -137,6 +180,7 @@ uv run python cli.py alpha list --alpha-module path/to/my_alpha.py # include a | `--output-dir` | `alphas` | Output directory | | `--lookback` | `5` | Lookback days (passed to alphas that accept it) | | `--vol-window` | `20` | Volatility window (passed to alphas that accept it) | +| `--feature-path` | — | Daily derived/feature parquet file or dataset to left-join on `symbol_id,date`; repeatable | | `--alpha-module` | — | External module(s) to import first; repeatable. Dotted path or `.py` file | | `--param` | — | Extra constructor param as `name=value`; repeatable | @@ -370,6 +414,8 @@ between phases (data is stored long/tidy): OHLC scale under `qfq`/`hfq`; `turn` is turnover %, `pctChg` daily % change, `tradestatus`/`isST` are 0/1 flags, and `peTTM`/`pbMRQ`/`psTTM`/`pcfNcfTTM` are baostock valuation ratios.) +- **derived** (`DERIVED_KEY_COLUMNS` + values): required keys `symbol_id, date`; + value columns are user/plugin-defined and must be numeric in v1. - **alpha** (`ALPHA_COLUMNS`): `symbol_id, date, alpha_name, weight` - **combo** (`COMBO_COLUMNS`): `symbol_id, date, combo_name, weight` - **portfolio positions** (`POSITION_COLUMNS`): `symbol_id, date, portfolio_name, target_weight, target_value, target_shares, position_shares, position_value, price` @@ -387,8 +433,11 @@ directory yields an extra `month` (`YYYY-MM`) partition column on top of - `cli.py` — entry point wiring the file-based phases together - `pipeline/data/` — universe resolution + download → `data/daily_bars/{universe}/month=YYYY-MM/*.pq` +- `pipeline/derived/` — daily derived-data ingestion, validation, plugin registry, + and built-in derived computations → `derived/*.pq` - `pipeline/alpha/` — `base.py` (`BaseAlpha`), `registry.py` (factory + plugin loader), `library/` (built-in alphas), `compute.py` (`compute_alpha` / `evaluate_alpha`) +- `pipeline/features/` — compatibility wrappers for the derived-data registry - `pipeline/combo/` — alpha combination → `combos/*.pq` - `pipeline/portfolio/` — construction, A-share lot/limit rules, constraints, reference next-open simulator, and research metrics @@ -410,6 +459,9 @@ constructed positions, fills/costs, P&L, and target-weight research metrics. - [x] **Reference execution simulation** — next-open fills over constructed `position_shares`, with suspension, price-limit, volume-cap, transaction-cost, and slippage controls. +- [x] **Derived/custom daily data ("Level 2")** — ingest user CSV/parquet files + or compute plugin outputs as validated numeric daily datasets under + `derived/{name}.pq`; alpha joins continue through `--feature-path`. - [ ] **Optional Backtrader adapter** — Backtrader is available as the `backtrader` extra for possible future event-driven/broker-style experiments, but it is not part of the current canonical portfolio workflow. @@ -419,3 +471,29 @@ constructed positions, fills/costs, P&L, and target-weight research metrics. and intraday VWAP. These need a tick / L1–L2 quote feed (typically a paid or brokerage data tier); the free daily sources here only expose daily bars, so this is a separate data phase rather than extra columns on the daily schema. + +### Additional TODOs + +The following items are intended extensions beyond the current daily +alpha-to-portfolio pipeline: + +- **Long-only portfolio mode** — add a construction option that converts + alpha/combo weights into a long-only book while preserving existing lot, + price, suspension, and volume-cap handling. +- **Index-short hedging mode** — support portfolios that hold long A-share + names while shorting an index or index proxy for market exposure control. +- **Expanded universe presets** — add explicit universe aliases for CSI 300, + CSI 500, CSI 1000, and CSI 1800, while keeping file-based and comma-separated + custom universes available. +- **Categorical derived data** — extend the numeric-only derived-data v1 contract + to support categorical inputs such as industry classifications. In this + project, "Level 2" means customized second-level research data produced by + users or plugins; it does not necessarily mean exchange order-book/L2 quote + feeds. +- **Minute bar data** — continue extending the raw minute-bar and feature + workflow. The initial Baostock 5-minute download and daily feature plugin path + exist; intraday execution and replacing canonical daily bars remain out of + scope unless explicitly added later. +- **Industry data** — add industry classification inputs for filtering, + grouping, exposure reporting, neutralization, or industry-aware portfolio + construction. diff --git a/cli.py b/cli.py index 0078e34..a6ed9e2 100644 --- a/cli.py +++ b/cli.py @@ -3,6 +3,7 @@ Phases: data — Download daily bars to parquet + derived — Ingest or compute daily derived data alpha — Compute alpha weights from data feature — Compute daily features from minute bars combo — Combine alphas into a single weight @@ -14,6 +15,7 @@ import logging import click from pipeline.data.cli import data +from pipeline.derived.cli import derived from pipeline.alpha.cli import alpha from pipeline.features.cli import feature from pipeline.combo.cli import combo @@ -41,6 +43,7 @@ def cli(log_level): cli.add_command(data) +cli.add_command(derived) cli.add_command(alpha) cli.add_command(feature) cli.add_command(combo) diff --git a/docs/minute_bar_data.md b/docs/minute_bar_data.md index ba1a139..38adf85 100644 --- a/docs/minute_bar_data.md +++ b/docs/minute_bar_data.md @@ -16,17 +16,20 @@ The default layout is: data/minute_bars/{universe}/frequency=5m/month=YYYY-MM/*.pq ``` -Feature plugins can aggregate those bars to daily `symbol_id,date` feature +Derived-data plugins can aggregate those bars to daily `symbol_id,date` numeric files, for example: ```bash -uv run python cli.py feature compute \ +uv run python cli.py derived compute \ --minute-path data/minute_bars/sh600000 \ --daily-path data/daily_bars/sh600000 \ - --feature-type minute_daily_summary \ - --feature-name minute_summary + --derived-type minute_daily_summary \ + --derived-name minute_summary ``` +The legacy `feature compute` command delegates to the same derived-data +registry and remains available for existing scripts. + ## Daily vs Minute Reconciliation Baostock's daily raw bars and 5-minute raw bars are close, but they should not diff --git a/pipeline/alpha/cli.py b/pipeline/alpha/cli.py index c01b54f..4590c7a 100644 --- a/pipeline/alpha/cli.py +++ b/pipeline/alpha/cli.py @@ -58,7 +58,7 @@ def list_(alpha_modules): @click.option("--vol-window", default=20, type=int, help="Volatility window (reversal_vol only)") @click.option( "--feature-path", "feature_paths", multiple=True, - help="Daily feature parquet file/dataset to left-join on symbol_id,date (repeatable)", + help="Daily derived/feature parquet file or dataset to left-join on symbol_id,date (repeatable)", ) @click.option( "--alpha-module", "alpha_modules", multiple=True, diff --git a/pipeline/alpha/compute.py b/pipeline/alpha/compute.py index 857a9cd..a5cc211 100644 --- a/pipeline/alpha/compute.py +++ b/pipeline/alpha/compute.py @@ -15,7 +15,11 @@ import pandas as pd from pipeline.alpha.registry import get_alpha from pipeline.common.schema import ALPHA_COLUMNS -from pipeline.features.compute import FEATURE_KEY_COLUMNS, read_feature_frames, validate_feature_frame +from pipeline.derived.compute import ( + DERIVED_KEY_COLUMNS, + read_derived_frames, + validate_derived_frame, +) logger = logging.getLogger(__name__) @@ -40,15 +44,15 @@ def join_feature_frames( data: pd.DataFrame, feature_frames: Iterable[pd.DataFrame], ) -> pd.DataFrame: - """Left-join validated daily feature frames onto long daily data.""" + """Left-join validated daily derived/feature frames onto long daily data.""" out = data.copy() out["date"] = pd.to_datetime(out["date"]) existing = set(out.columns) joined_cols: list[str] = [] for frame in feature_frames: - features = validate_feature_frame(frame) - feature_cols = [col for col in features.columns if col not in FEATURE_KEY_COLUMNS] + features = validate_derived_frame(frame) + feature_cols = [col for col in features.columns if col not in DERIVED_KEY_COLUMNS] overlap = sorted(existing.intersection(feature_cols)) if overlap: raise ValueError( @@ -56,7 +60,7 @@ def join_feature_frames( ) out = out.merge( features, - on=FEATURE_KEY_COLUMNS, + on=DERIVED_KEY_COLUMNS, how="left", validate="many_to_one", ) @@ -171,7 +175,7 @@ def compute_alpha( """ feature_inputs: list[pd.DataFrame] = [] if feature_paths: - feature_inputs.extend(read_feature_frames(feature_paths)) + feature_inputs.extend(read_derived_frames(feature_paths)) if feature_frames: feature_inputs.extend(feature_frames) if feature_inputs: diff --git a/pipeline/common/schema.py b/pipeline/common/schema.py index 4857179..0ee8a15 100644 --- a/pipeline/common/schema.py +++ b/pipeline/common/schema.py @@ -44,6 +44,13 @@ MINUTE_BAR_COLUMNS: Final[list[str]] = [ "adjustflag", # str: baostock adjustment flag; '3' for raw/unadjusted ] +# Required key columns for daily derived-data parquet files. Value columns are +# user/plugin-defined and must be numeric. +DERIVED_KEY_COLUMNS: Final[list[str]] = [ + "symbol_id", # str + "date", # date: normalized daily timestamp +] + # Required columns for alpha parquet files. # Alphas are position WEIGHTS: positive=long, negative=short. ALPHA_COLUMNS: Final[list[str]] = [ diff --git a/pipeline/derived/__init__.py b/pipeline/derived/__init__.py new file mode 100644 index 0000000..012b7ea --- /dev/null +++ b/pipeline/derived/__init__.py @@ -0,0 +1,2 @@ +"""Daily derived-data plugin package.""" + diff --git a/pipeline/derived/base.py b/pipeline/derived/base.py new file mode 100644 index 0000000..23faeca --- /dev/null +++ b/pipeline/derived/base.py @@ -0,0 +1,38 @@ +"""Base class for daily derived-data plugins.""" + +from abc import ABC, abstractmethod + +import pandas as pd + + +class BaseDerivedData(ABC): + """Compute daily, symbol-keyed numeric derived data. + + Derived-data plugins may use daily bars, minute bars, or both as inputs, but + they must always return daily rows keyed by ``symbol_id,date``. + """ + + #: Unique registry key. Every concrete derived-data plugin must set this. + name: str = "" + + @abstractmethod + def compute( + self, + daily: pd.DataFrame | None = None, + minute: pd.DataFrame | None = None, + ) -> pd.DataFrame: + """Compute daily derived data. + + Args: + daily: Optional daily market data. + minute: Optional raw minute bars. + + Returns: + DataFrame with ``symbol_id``, ``date``, and one or more numeric + derived-data columns. + """ + + def __repr__(self) -> str: + params = ", ".join(f"{k}={v!r}" for k, v in vars(self).items()) + return f"{type(self).__name__}({params})" + diff --git a/pipeline/derived/cli.py b/pipeline/derived/cli.py new file mode 100644 index 0000000..4f9cb4a --- /dev/null +++ b/pipeline/derived/cli.py @@ -0,0 +1,145 @@ +"""CLI for daily derived-data ingestion and computation.""" + +import click +import pandas as pd + +from pipeline.derived.compute import ( + DERIVED_KEY_COLUMNS, + compute_derived, + read_derived_frame, + write_derived_frame, +) +from pipeline.derived.registry import ( + available_derived, + load_derived_module, +) + + +@click.group(name="derived") +def derived(): + """Ingest, compute, and validate daily derived data.""" + + +def _coerce(value: str): + """Best-effort coercion of a CLI string to int, then float, else str.""" + for cast in (int, float): + try: + return cast(value) + except ValueError: + continue + return value + + +def _parse_params(pairs: tuple[str, ...]) -> dict: + """Parse repeated ``name=value`` options into a params dict.""" + params: dict = {} + for pair in pairs: + if "=" not in pair: + raise click.BadParameter(f"--param must be name=value, got '{pair}'") + key, value = pair.split("=", 1) + params[key.strip()] = _coerce(value.strip()) + return params + + +def _read_optional_parquet(path: str | None) -> pd.DataFrame | None: + return None if path is None else pd.read_parquet(path) + + +def _summarize(result: pd.DataFrame) -> str: + value_cols = [col for col in result.columns if col not in DERIVED_KEY_COLUMNS] + return f"{len(result):,} rows, {len(value_cols)} columns" + + +@derived.command("list") +@click.option( + "--derived-module", "derived_modules", multiple=True, + help="External module(s) to import first (dotted path or .py file)", +) +def list_(derived_modules): + """List the registered derived-data plugin types.""" + for spec in derived_modules: + load_derived_module(spec) + for name in available_derived(): + click.echo(name) + + +@derived.command("validate") +@click.option("--input-path", required=True, help="CSV/parquet file or parquet dataset to validate") +def validate(input_path): + """Validate a daily derived-data file without writing output.""" + try: + result = read_derived_frame(input_path) + except Exception as exc: + raise click.ClickException(str(exc)) from exc + click.echo(f"Valid derived data: {input_path} ({_summarize(result)})") + + +@derived.command("ingest") +@click.option("--input-path", required=True, help="CSV/parquet file to ingest") +@click.option("--derived-name", required=True, help="Name for this derived-data output file") +@click.option("--output-dir", default="derived", help="Directory to save derived parquet") +def ingest(input_path, derived_name, output_dir): + """Ingest a user-provided daily derived-data CSV/parquet file.""" + try: + result = read_derived_frame(input_path) + out_path = write_derived_frame(result, derived_name, output_dir=output_dir) + except Exception as exc: + raise click.ClickException(str(exc)) from exc + click.echo(f"Saved derived data: {out_path} ({_summarize(result)})") + + +@derived.command("compute") +@click.option("--daily-path", default=None, help="Optional daily data parquet/dataset") +@click.option("--minute-path", default=None, help="Optional minute parquet/dataset") +@click.option("--derived-type", required=True, help="Registry key of the derived-data plugin") +@click.option("--derived-name", required=True, help="Name for this derived-data output file") +@click.option("--output-dir", default="derived", help="Directory to save derived parquet") +@click.option( + "--derived-module", "derived_modules", multiple=True, + help="External module(s) to import so their derived-data plugins register", +) +@click.option( + "--param", "extra_params", multiple=True, + help="Extra derived-data constructor param as name=value (repeatable)", +) +def compute( + daily_path, + minute_path, + derived_type, + derived_name, + output_dir, + derived_modules, + extra_params, +): + """Compute one daily derived-data file from daily and/or minute inputs.""" + for spec in derived_modules: + load_derived_module(spec) + + options = available_derived() + if derived_type not in options: + raise click.BadParameter( + f"Unknown derived-type '{derived_type}'. Available: {options}. " + f"Use --derived-module to register an external derived-data plugin.", + param_hint="--derived-type", + ) + if daily_path is None and minute_path is None: + raise click.UsageError("At least one of --daily-path or --minute-path is required") + + daily = _read_optional_parquet(daily_path) + if daily_path: + click.echo(f"Loaded daily data: {len(daily):,} rows from {daily_path}") + minute = _read_optional_parquet(minute_path) + if minute_path: + click.echo(f"Loaded minute bars: {len(minute):,} rows from {minute_path}") + + try: + result = compute_derived( + derived_type=derived_type, + daily=daily, + minute=minute, + **_parse_params(extra_params), + ) + out_path = write_derived_frame(result, derived_name, output_dir=output_dir) + except Exception as exc: + raise click.ClickException(str(exc)) from exc + click.echo(f"Saved derived data: {out_path} ({_summarize(result)})") diff --git a/pipeline/derived/compute.py b/pipeline/derived/compute.py new file mode 100644 index 0000000..8728869 --- /dev/null +++ b/pipeline/derived/compute.py @@ -0,0 +1,115 @@ +"""Derived-data computation and validation.""" + +import csv +import logging +from pathlib import Path +from typing import Iterable + +import pandas as pd +from pandas.api.types import is_bool_dtype, is_numeric_dtype + +from pipeline.common.schema import DERIVED_KEY_COLUMNS +from pipeline.derived.registry import get_derived + +logger = logging.getLogger(__name__) + + +def validate_derived_frame(derived: pd.DataFrame) -> pd.DataFrame: + """Validate and normalize a daily derived-data frame. + + A valid derived frame is keyed by unique ``symbol_id,date`` rows and has at + least one numeric value column beyond those keys. Dates are normalized to + daily timestamps before duplicate-key checks. + """ + duplicated = derived.columns[derived.columns.duplicated()].tolist() + if duplicated: + raise ValueError(f"Derived data has duplicate columns: {duplicated}") + + missing = [col for col in DERIVED_KEY_COLUMNS if col not in derived.columns] + if missing: + raise ValueError(f"Derived data missing required columns: {missing}") + + out = derived.copy() + out["date"] = pd.to_datetime(out["date"]).dt.normalize() + + if out.duplicated(DERIVED_KEY_COLUMNS).any(): + raise ValueError("Derived data has duplicate symbol_id,date rows") + + value_cols = [col for col in out.columns if col not in DERIVED_KEY_COLUMNS] + if not value_cols: + raise ValueError("Derived data must include at least one value column") + + non_numeric = [ + col + for col in value_cols + if is_bool_dtype(out[col]) or not is_numeric_dtype(out[col]) + ] + if non_numeric: + raise ValueError(f"Derived data value columns must be numeric: {non_numeric}") + + out = out[DERIVED_KEY_COLUMNS + value_cols].copy() + return out.sort_values(DERIVED_KEY_COLUMNS).reset_index(drop=True) + + +def compute_derived( + derived_type: str, + daily: pd.DataFrame | None = None, + minute: pd.DataFrame | None = None, + **params, +) -> pd.DataFrame: + """Compute one registered derived-data plugin.""" + if daily is None and minute is None: + raise ValueError("Derived data computation requires --daily-path or --minute-path") + + derived = get_derived(derived_type, **params) + result = validate_derived_frame(derived.compute(daily=daily, minute=minute)) + value_cols = [col for col in result.columns if col not in DERIVED_KEY_COLUMNS] + logger.info( + "Derived data '%s' (%r): %d symbols × %d dates, columns=%s", + derived_type, + derived, + result["symbol_id"].nunique(), + result["date"].nunique(), + value_cols, + ) + return result + + +def read_derived_frame(path: str | Path) -> pd.DataFrame: + """Read and validate one derived CSV/parquet file or parquet dataset.""" + path = Path(path) + if path.suffix.lower() == ".csv": + return validate_derived_frame(_read_csv_with_duplicate_header_check(path)) + return validate_derived_frame(pd.read_parquet(path)) + + +def read_derived_frames(derived_paths: Iterable[str | Path]) -> list[pd.DataFrame]: + """Read and validate derived-data files.""" + return [read_derived_frame(path) for path in derived_paths] + + +def write_derived_frame( + derived: pd.DataFrame, + derived_name: str, + output_dir: str | Path = "derived", +) -> Path: + """Validate and write derived data to ``{output_dir}/{derived_name}.pq``.""" + result = validate_derived_frame(derived) + output_dir = Path(output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + out_path = output_dir / f"{derived_name}.pq" + result.to_parquet(out_path, index=False) + return out_path + + +def _read_csv_with_duplicate_header_check(path: Path) -> pd.DataFrame: + with path.open(newline="") as fh: + reader = csv.reader(fh) + try: + header = next(reader) + except StopIteration as exc: + raise ValueError("CSV input is empty") from exc + duplicated = sorted({col for col in header if header.count(col) > 1}) + if duplicated: + raise ValueError(f"Derived data has duplicate columns: {duplicated}") + return pd.read_csv(path) diff --git a/pipeline/derived/library/__init__.py b/pipeline/derived/library/__init__.py new file mode 100644 index 0000000..c4b0e2e --- /dev/null +++ b/pipeline/derived/library/__init__.py @@ -0,0 +1,4 @@ +"""Built-in derived-data library.""" + +from pipeline.derived.library import minute_daily_summary # noqa: F401 + diff --git a/pipeline/derived/library/minute_daily_summary.py b/pipeline/derived/library/minute_daily_summary.py new file mode 100644 index 0000000..290032e --- /dev/null +++ b/pipeline/derived/library/minute_daily_summary.py @@ -0,0 +1,88 @@ +"""Daily summary data derived from raw minute bars.""" + +import numpy as np +import pandas as pd + +from pipeline.derived.base import BaseDerivedData +from pipeline.derived.registry import register_derived + + +@register_derived +class MinuteDailySummaryDerived(BaseDerivedData): + """Aggregate intraday bars into daily summary columns.""" + + name = "minute_daily_summary" + + def compute( + self, + daily: pd.DataFrame | None = None, + minute: pd.DataFrame | None = None, + ) -> pd.DataFrame: + if minute is None: + raise ValueError("minute_daily_summary requires minute input") + + minute = minute.copy() + minute["date"] = pd.to_datetime(minute["date"]).dt.normalize() + sort_cols = ["symbol_id", "date"] + if "datetime" in minute.columns: + minute["datetime"] = pd.to_datetime(minute["datetime"]) + sort_cols.append("datetime") + elif "time" in minute.columns: + sort_cols.append("time") + minute = minute.sort_values(sort_cols) + + grouped = minute.groupby(["symbol_id", "date"], sort=True) + summary = grouped.agg( + minute_bar_count=("close", "count"), + first_open=("open", "first"), + last_close=("close", "last"), + high=("high", "max"), + low=("low", "min"), + volume_sum=("volume", "sum"), + amount_sum=("amount", "sum"), + ) + summary["minute_intraday_return"] = ( + summary["last_close"] / summary["first_open"] - 1.0 + ) + summary["minute_intraday_range"] = summary["high"] / summary["low"] - 1.0 + summary["minute_vwap"] = ( + summary["amount_sum"] / summary["volume_sum"].where(summary["volume_sum"] > 0) + ) + summary = summary.reset_index() + + if daily is not None: + daily_keys = daily[["symbol_id", "date"]].copy() + daily_keys["date"] = pd.to_datetime(daily_keys["date"]).dt.normalize() + daily_keys = daily_keys.drop_duplicates(["symbol_id", "date"]) + result = daily_keys.merge(summary, on=["symbol_id", "date"], how="left") + if "close" in daily.columns: + daily_close = daily[["symbol_id", "date", "close"]].copy() + daily_close["date"] = pd.to_datetime(daily_close["date"]).dt.normalize() + daily_close = daily_close.drop_duplicates(["symbol_id", "date"]) + result = result.merge( + daily_close.rename(columns={"close": "daily_close"}), + on=["symbol_id", "date"], + how="left", + ) + reference_close = result["daily_close"].fillna(result["last_close"]) + else: + reference_close = result["last_close"] + else: + result = summary + reference_close = result["last_close"] + + result["minute_vwap_deviation"] = ( + result["minute_vwap"] / reference_close.replace(0.0, np.nan) - 1.0 + ) + return result[ + [ + "symbol_id", + "date", + "minute_bar_count", + "minute_intraday_return", + "minute_intraday_range", + "minute_vwap", + "minute_vwap_deviation", + ] + ] + diff --git a/pipeline/derived/registry.py b/pipeline/derived/registry.py new file mode 100644 index 0000000..e924900 --- /dev/null +++ b/pipeline/derived/registry.py @@ -0,0 +1,80 @@ +"""Registry and factory for daily derived-data plugins.""" + +import importlib +import importlib.util +import inspect +from pathlib import Path +from typing import Optional, Type + +from pipeline.derived.base import BaseDerivedData + +_REGISTRY: dict[str, Type[BaseDerivedData]] = {} +_builtins_loaded = False + + +def register_derived(cls: Type[BaseDerivedData]) -> Type[BaseDerivedData]: + """Class decorator that registers derived data under ``BaseDerivedData.name``.""" + if not (isinstance(cls, type) and issubclass(cls, BaseDerivedData)): + raise TypeError(f"{cls!r} is not a BaseDerivedData subclass") + key = getattr(cls, "name", "") + if not key: + raise ValueError(f"{cls.__name__} must set a non-empty class attribute `name`") + existing = _REGISTRY.get(key) + if existing is not None and existing is not cls: + raise ValueError( + f"Derived data name '{key}' already registered by {existing.__name__}" + ) + _REGISTRY[key] = cls + return cls + + +def available_derived() -> list[str]: + """Sorted names of all registered derived-data plugins.""" + _ensure_builtins() + return sorted(_REGISTRY) + + +def get_derived(name: str, **params) -> BaseDerivedData: + """Instantiate a registered derived-data plugin by name. + + Only parameters accepted by the plugin class's ``__init__`` are forwarded. + """ + _ensure_builtins() + if name not in _REGISTRY: + raise KeyError(f"Unknown derived data '{name}'. Available: {sorted(_REGISTRY)}") + cls = _REGISTRY[name] + accepted = _accepted_params(cls) + kwargs = params if accepted is None else {k: v for k, v in params.items() if k in accepted} + return cls(**kwargs) + + +def load_derived_module(spec: str) -> None: + """Import an external module so its ``@register_derived`` classes register.""" + looks_like_file = spec.endswith(".py") or Path(spec).expanduser().exists() + if looks_like_file: + path = Path(spec).expanduser().resolve() + if not path.exists(): + raise FileNotFoundError(f"Derived data module not found: {path}") + module_spec = importlib.util.spec_from_file_location(path.stem, path) + if module_spec is None or module_spec.loader is None: + raise ImportError(f"Cannot load derived data module from {path}") + module = importlib.util.module_from_spec(module_spec) + module_spec.loader.exec_module(module) + else: + importlib.import_module(spec) + + +def _accepted_params(cls: Type[BaseDerivedData]) -> Optional[set[str]]: + """Param names ``cls.__init__`` accepts, or None if it takes ``**kwargs``.""" + sig = inspect.signature(cls.__init__) + if any(p.kind is p.VAR_KEYWORD for p in sig.parameters.values()): + return None + return {name for name in sig.parameters if name != "self"} + + +def _ensure_builtins() -> None: + global _builtins_loaded + if not _builtins_loaded: + import pipeline.derived.library # noqa: F401 + _builtins_loaded = True + diff --git a/pipeline/features/base.py b/pipeline/features/base.py index 55a8697..c3d989b 100644 --- a/pipeline/features/base.py +++ b/pipeline/features/base.py @@ -1,34 +1,8 @@ -"""Base class for daily feature plugins.""" +"""Compatibility alias for daily feature plugins. -from abc import ABC, abstractmethod +The canonical plugin API is ``pipeline.derived``. ``BaseFeature`` remains as an +alias so existing external feature modules continue to register unchanged. +""" -import pandas as pd +from pipeline.derived.base import BaseDerivedData as BaseFeature - -class BaseFeature(ABC): - """Aggregate raw minute bars into daily, symbol-keyed feature columns.""" - - #: Unique registry key. Every concrete feature must set this to a non-empty str. - name: str = "" - - @abstractmethod - def compute( - self, - minute: pd.DataFrame, - daily: pd.DataFrame | None = None, - ) -> pd.DataFrame: - """Compute daily features. - - Args: - minute: Raw minute bars with ``symbol_id`` and ``date`` keys. - daily: Optional daily data frame for calendar alignment or - reference daily columns. - - Returns: - DataFrame with ``symbol_id``, ``date``, and one or more numeric - feature columns. - """ - - def __repr__(self) -> str: - params = ", ".join(f"{k}={v!r}" for k, v in vars(self).items()) - return f"{type(self).__name__}({params})" diff --git a/pipeline/features/compute.py b/pipeline/features/compute.py index 2394745..ef5374a 100644 --- a/pipeline/features/compute.py +++ b/pipeline/features/compute.py @@ -1,49 +1,23 @@ -"""Feature computation and validation.""" +"""Compatibility wrappers for daily feature computation and validation.""" -import logging from pathlib import Path from typing import Iterable import pandas as pd -from pandas.api.types import is_numeric_dtype -from pipeline.features.registry import get_feature +from pipeline.derived.compute import ( + DERIVED_KEY_COLUMNS, + compute_derived, + read_derived_frames, + validate_derived_frame, +) -logger = logging.getLogger(__name__) - -FEATURE_KEY_COLUMNS = ["symbol_id", "date"] +FEATURE_KEY_COLUMNS = DERIVED_KEY_COLUMNS def validate_feature_frame(features: pd.DataFrame) -> pd.DataFrame: - """Validate and normalize a daily feature frame. - - A valid feature frame is keyed by unique ``symbol_id,date`` rows and has at - least one numeric feature column beyond those keys. - """ - duplicated = features.columns[features.columns.duplicated()].tolist() - if duplicated: - raise ValueError(f"Feature output has duplicate columns: {duplicated}") - - missing = [col for col in FEATURE_KEY_COLUMNS if col not in features.columns] - if missing: - raise ValueError(f"Feature output missing required columns: {missing}") - - out = features.copy() - out["date"] = pd.to_datetime(out["date"]) - - if out.duplicated(FEATURE_KEY_COLUMNS).any(): - raise ValueError("Feature output has duplicate symbol_id,date rows") - - feature_cols = [col for col in out.columns if col not in FEATURE_KEY_COLUMNS] - if not feature_cols: - raise ValueError("Feature output must include at least one feature column") - - non_numeric = [col for col in feature_cols if not is_numeric_dtype(out[col])] - if non_numeric: - raise ValueError(f"Feature columns must be numeric: {non_numeric}") - - out = out[FEATURE_KEY_COLUMNS + feature_cols].copy() - return out.sort_values(FEATURE_KEY_COLUMNS).reset_index(drop=True) + """Validate and normalize a legacy daily feature frame.""" + return validate_derived_frame(features) def compute_feature( @@ -52,24 +26,16 @@ def compute_feature( daily: pd.DataFrame | None = None, **params, ) -> pd.DataFrame: - """Compute one registered feature from raw minute bars.""" - feature = get_feature(feature_type, **params) - result = validate_feature_frame(feature.compute(minute=minute, daily=daily)) - feature_cols = [col for col in result.columns if col not in FEATURE_KEY_COLUMNS] - logger.info( - "Feature '%s' (%r): %d symbols × %d dates, columns=%s", - feature_type, - feature, - result["symbol_id"].nunique(), - result["date"].nunique(), - feature_cols, + """Compute one registered feature through the derived-data registry.""" + return compute_derived( + derived_type=feature_type, + daily=daily, + minute=minute, + **params, ) - return result def read_feature_frames(feature_paths: Iterable[str | Path]) -> list[pd.DataFrame]: - """Read and validate feature parquet files.""" - return [ - validate_feature_frame(pd.read_parquet(path)) - for path in feature_paths - ] + """Read and validate feature/derived-data parquet files.""" + return read_derived_frames(feature_paths) + diff --git a/pipeline/features/library/minute_daily_summary.py b/pipeline/features/library/minute_daily_summary.py index 75b3b7e..7120970 100644 --- a/pipeline/features/library/minute_daily_summary.py +++ b/pipeline/features/library/minute_daily_summary.py @@ -1,84 +1,16 @@ -"""Daily summary features derived from raw minute bars.""" +"""Compatibility wrapper for the built-in minute daily summary plugin.""" -import numpy as np import pandas as pd -from pipeline.features.base import BaseFeature -from pipeline.features.registry import register_feature +from pipeline.derived.library.minute_daily_summary import MinuteDailySummaryDerived -@register_feature -class MinuteDailySummaryFeature(BaseFeature): - """Aggregate intraday bars into daily summary columns.""" - - name = "minute_daily_summary" +class MinuteDailySummaryFeature(MinuteDailySummaryDerived): + """Legacy minute-first wrapper around the derived-data implementation.""" def compute( self, minute: pd.DataFrame, daily: pd.DataFrame | None = None, ) -> pd.DataFrame: - minute = minute.copy() - minute["date"] = pd.to_datetime(minute["date"]) - sort_cols = ["symbol_id", "date"] - if "datetime" in minute.columns: - minute["datetime"] = pd.to_datetime(minute["datetime"]) - sort_cols.append("datetime") - elif "time" in minute.columns: - sort_cols.append("time") - minute = minute.sort_values(sort_cols) - - grouped = minute.groupby(["symbol_id", "date"], sort=True) - summary = grouped.agg( - minute_bar_count=("close", "count"), - first_open=("open", "first"), - last_close=("close", "last"), - high=("high", "max"), - low=("low", "min"), - volume_sum=("volume", "sum"), - amount_sum=("amount", "sum"), - ) - summary["minute_intraday_return"] = ( - summary["last_close"] / summary["first_open"] - 1.0 - ) - summary["minute_intraday_range"] = summary["high"] / summary["low"] - 1.0 - summary["minute_vwap"] = ( - summary["amount_sum"] / summary["volume_sum"].where(summary["volume_sum"] > 0) - ) - summary = summary.reset_index() - - if daily is not None: - daily_keys = daily[["symbol_id", "date"]].copy() - daily_keys["date"] = pd.to_datetime(daily_keys["date"]) - daily_keys = daily_keys.drop_duplicates(["symbol_id", "date"]) - result = daily_keys.merge(summary, on=["symbol_id", "date"], how="left") - if "close" in daily.columns: - daily_close = daily[["symbol_id", "date", "close"]].copy() - daily_close["date"] = pd.to_datetime(daily_close["date"]) - daily_close = daily_close.drop_duplicates(["symbol_id", "date"]) - result = result.merge( - daily_close.rename(columns={"close": "daily_close"}), - on=["symbol_id", "date"], - how="left", - ) - reference_close = result["daily_close"].fillna(result["last_close"]) - else: - reference_close = result["last_close"] - else: - result = summary - reference_close = result["last_close"] - - result["minute_vwap_deviation"] = ( - result["minute_vwap"] / reference_close.replace(0.0, np.nan) - 1.0 - ) - return result[ - [ - "symbol_id", - "date", - "minute_bar_count", - "minute_intraday_return", - "minute_intraday_range", - "minute_vwap", - "minute_vwap_deviation", - ] - ] + return super().compute(daily=daily, minute=minute) diff --git a/pipeline/features/registry.py b/pipeline/features/registry.py index b026dfc..92cf4d5 100644 --- a/pipeline/features/registry.py +++ b/pipeline/features/registry.py @@ -1,79 +1,36 @@ -"""Registry and factory for daily feature plugins.""" +"""Compatibility registry wrappers for daily feature plugins.""" -import importlib -import importlib.util -import inspect -from pathlib import Path -from typing import Optional, Type +from typing import Type +from pipeline.derived.base import BaseDerivedData +from pipeline.derived.registry import ( + available_derived, + get_derived, + load_derived_module, + register_derived, +) from pipeline.features.base import BaseFeature -_REGISTRY: dict[str, Type[BaseFeature]] = {} -_builtins_loaded = False - def register_feature(cls: Type[BaseFeature]) -> Type[BaseFeature]: - """Class decorator that registers a feature under ``BaseFeature.name``.""" - if not (isinstance(cls, type) and issubclass(cls, BaseFeature)): - raise TypeError(f"{cls!r} is not a BaseFeature subclass") - key = getattr(cls, "name", "") - if not key: - raise ValueError(f"{cls.__name__} must set a non-empty class attribute `name`") - existing = _REGISTRY.get(key) - if existing is not None and existing is not cls: - raise ValueError( - f"Feature name '{key}' already registered by {existing.__name__}" - ) - _REGISTRY[key] = cls - return cls + """Register a legacy feature plugin in the derived-data registry.""" + return register_derived(cls) def available_features() -> list[str]: - """Sorted names of all registered features (built-ins are loaded lazily).""" - _ensure_builtins() - return sorted(_REGISTRY) + """Sorted names of all registered feature/derived-data plugins.""" + return available_derived() -def get_feature(name: str, **params) -> BaseFeature: - """Instantiate a registered feature by name. +def get_feature(name: str, **params) -> BaseDerivedData: + """Instantiate a registered feature/derived-data plugin by name.""" + if name == "minute_daily_summary": + from pipeline.features.library.minute_daily_summary import MinuteDailySummaryFeature - Only parameters accepted by the feature class's ``__init__`` are forwarded. - """ - _ensure_builtins() - if name not in _REGISTRY: - raise KeyError(f"Unknown feature '{name}'. Available: {sorted(_REGISTRY)}") - cls = _REGISTRY[name] - accepted = _accepted_params(cls) - kwargs = params if accepted is None else {k: v for k, v in params.items() if k in accepted} - return cls(**kwargs) + return MinuteDailySummaryFeature(**params) + return get_derived(name, **params) def load_feature_module(spec: str) -> None: """Import an external module so its ``@register_feature`` classes register.""" - looks_like_file = spec.endswith(".py") or Path(spec).expanduser().exists() - if looks_like_file: - path = Path(spec).expanduser().resolve() - if not path.exists(): - raise FileNotFoundError(f"Feature module not found: {path}") - module_spec = importlib.util.spec_from_file_location(path.stem, path) - if module_spec is None or module_spec.loader is None: - raise ImportError(f"Cannot load feature module from {path}") - module = importlib.util.module_from_spec(module_spec) - module_spec.loader.exec_module(module) - else: - importlib.import_module(spec) - - -def _accepted_params(cls: Type[BaseFeature]) -> Optional[set[str]]: - """Param names ``cls.__init__`` accepts, or None if it takes ``**kwargs``.""" - sig = inspect.signature(cls.__init__) - if any(p.kind is p.VAR_KEYWORD for p in sig.parameters.values()): - return None - return {name for name in sig.parameters if name != "self"} - - -def _ensure_builtins() -> None: - global _builtins_loaded - if not _builtins_loaded: - import pipeline.features.library # noqa: F401 - _builtins_loaded = True + load_derived_module(spec) diff --git a/tests/test_derived.py b/tests/test_derived.py new file mode 100644 index 0000000..5e20223 --- /dev/null +++ b/tests/test_derived.py @@ -0,0 +1,261 @@ +"""Tests for daily derived-data ingestion and plugins.""" + +import textwrap + +import numpy as np +import pandas as pd +import pytest +from click.testing import CliRunner + +from cli import cli +from pipeline.alpha.compute import join_feature_frames +from pipeline.derived.compute import compute_derived, validate_derived_frame +from pipeline.derived.registry import available_derived, get_derived, load_derived_module + + +def _daily_bars() -> pd.DataFrame: + return pd.DataFrame({ + "symbol_id": ["sh600000", "sz000001", "sh600000"], + "date": pd.to_datetime(["2024-01-02", "2024-01-02", "2024-01-03"]), + "open": [10.0, 20.0, 11.0], + "close": [10.5, 20.5, 11.5], + "volume": [1000.0, 2000.0, 1200.0], + }) + + +def _minute_bars() -> pd.DataFrame: + return pd.DataFrame({ + "symbol_id": ["sh600000", "sh600000", "sz000001"], + "datetime": pd.to_datetime([ + "2024-01-02 09:35:00", + "2024-01-02 09:40:00", + "2024-01-02 09:35:00", + ]), + "date": pd.to_datetime(["2024-01-02", "2024-01-02", "2024-01-02"]), + "time": ["09:35:00", "09:40:00", "09:35:00"], + "open": [10.0, 10.5, 20.0], + "high": [11.0, 12.0, 21.0], + "low": [9.0, 10.0, 19.0], + "close": [10.5, 11.0, 20.5], + "volume": [100.0, 300.0, 200.0], + "amount": [1000.0, 3300.0, 4100.0], + }) + + +def test_validate_derived_frame_normalizes_and_sorts(): + result = validate_derived_frame(pd.DataFrame({ + "symbol_id": ["sz000001", "sh600000"], + "date": ["2024-01-02 15:00:00", "2024-01-02 09:30:00"], + "custom_value": [2.0, 1.0], + })) + + assert result["symbol_id"].tolist() == ["sh600000", "sz000001"] + assert result["date"].tolist() == [ + pd.Timestamp("2024-01-02"), + pd.Timestamp("2024-01-02"), + ] + + +def test_validate_derived_frame_rejects_missing_keys(): + with pytest.raises(ValueError, match="missing required"): + validate_derived_frame(pd.DataFrame({"symbol_id": ["sh600000"], "x": [1.0]})) + + +def test_validate_derived_frame_rejects_duplicate_normalized_keys(): + with pytest.raises(ValueError, match="duplicate symbol_id,date"): + validate_derived_frame(pd.DataFrame({ + "symbol_id": ["sh600000", "sh600000"], + "date": ["2024-01-02 09:30:00", "2024-01-02 15:00:00"], + "x": [1.0, 2.0], + })) + + +def test_validate_derived_frame_rejects_duplicate_columns(): + bad = pd.DataFrame( + [["sh600000", pd.Timestamp("2024-01-02"), 1.0, 2.0]], + columns=["symbol_id", "date", "dup", "dup"], + ) + with pytest.raises(ValueError, match="duplicate columns"): + validate_derived_frame(bad) + + +def test_validate_derived_frame_rejects_non_numeric_values(): + with pytest.raises(ValueError, match="numeric"): + validate_derived_frame(pd.DataFrame({ + "symbol_id": ["sh600000"], + "date": [pd.Timestamp("2024-01-02")], + "bad": ["not numeric"], + })) + + +def test_derived_ingest_cli_accepts_csv_and_parquet(tmp_path): + runner = CliRunner() + source = pd.DataFrame({ + "symbol_id": ["sz000001", "sh600000"], + "date": ["2024-01-02", "2024-01-02"], + "custom_value": [2.0, 1.0], + }) + csv_path = tmp_path / "custom.csv" + parquet_path = tmp_path / "custom.pq" + out_dir = tmp_path / "derived" + source.to_csv(csv_path, index=False) + source.to_parquet(parquet_path, index=False) + + csv_result = runner.invoke(cli, [ + "derived", "ingest", + "--input-path", str(csv_path), + "--derived-name", "csv_custom", + "--output-dir", str(out_dir), + ]) + assert csv_result.exit_code == 0, csv_result.output + + parquet_result = runner.invoke(cli, [ + "derived", "ingest", + "--input-path", str(parquet_path), + "--derived-name", "parquet_custom", + "--output-dir", str(out_dir), + ]) + assert parquet_result.exit_code == 0, parquet_result.output + + written = pd.read_parquet(out_dir / "csv_custom.pq") + assert written["symbol_id"].tolist() == ["sh600000", "sz000001"] + assert (out_dir / "parquet_custom.pq").exists() + + +def test_derived_validate_cli_rejects_duplicate_csv_columns(tmp_path): + runner = CliRunner() + csv_path = tmp_path / "bad.csv" + csv_path.write_text("symbol_id,date,x,x\nsh600000,2024-01-02,1.0,2.0\n") + + result = runner.invoke(cli, [ + "derived", "validate", + "--input-path", str(csv_path), + ]) + + assert result.exit_code != 0 + assert "duplicate columns" in result.output + + +def test_external_derived_plugin_loads_filters_params_and_uses_inputs(tmp_path): + module_path = tmp_path / "external_derived.py" + module_path.write_text(textwrap.dedent(''' + import pandas as pd + from pipeline.derived.base import BaseDerivedData + from pipeline.derived.registry import register_derived + + @register_derived + class FlexibleDerived(BaseDerivedData): + name = "flexible_derived_test" + + def __init__(self, scale: float = 1.0): + self.scale = scale + + def compute(self, daily=None, minute=None) -> pd.DataFrame: + result = None + if daily is not None: + result = daily[["symbol_id", "date", "close"]].copy() + result["daily_scaled_close"] = result.pop("close") * self.scale + if minute is not None: + minute_out = ( + minute.groupby(["symbol_id", "date"], as_index=False)["volume"] + .sum() + .rename(columns={"volume": "minute_volume_sum"}) + ) + minute_out["minute_volume_sum"] *= self.scale + result = minute_out if result is None else result.merge( + minute_out, on=["symbol_id", "date"], how="left" + ) + return result + ''')) + + load_derived_module(str(module_path)) + assert "flexible_derived_test" in available_derived() + + instance = get_derived("flexible_derived_test", scale=2.0, ignored=99) + assert instance.scale == 2.0 + assert not hasattr(instance, "ignored") + + daily_result = compute_derived( + "flexible_derived_test", + daily=_daily_bars(), + scale=2.0, + ignored=99, + ) + assert "daily_scaled_close" in daily_result.columns + assert np.isclose(daily_result["daily_scaled_close"].iloc[0], 21.0) + + minute_result = compute_derived( + "flexible_derived_test", + minute=_minute_bars(), + scale=2.0, + ) + assert "minute_volume_sum" in minute_result.columns + assert np.isclose( + minute_result.loc[minute_result["symbol_id"] == "sh600000", "minute_volume_sum"].iloc[0], + 800.0, + ) + + both_result = compute_derived( + "flexible_derived_test", + daily=_daily_bars(), + minute=_minute_bars(), + scale=1.0, + ) + assert {"daily_scaled_close", "minute_volume_sum"}.issubset(both_result.columns) + + +def test_derived_compute_cli_writes_builtin_minute_summary(tmp_path): + runner = CliRunner() + minute_path = tmp_path / "minute.pq" + out_dir = tmp_path / "derived" + _minute_bars().to_parquet(minute_path, index=False) + + result = runner.invoke(cli, [ + "derived", "compute", + "--minute-path", str(minute_path), + "--derived-type", "minute_daily_summary", + "--derived-name", "minute_summary", + "--output-dir", str(out_dir), + ]) + + assert result.exit_code == 0, result.output + written = pd.read_parquet(out_dir / "minute_summary.pq") + assert "minute_vwap" in written.columns + + +def test_alpha_feature_join_rejects_derived_column_collisions(): + data = _daily_bars() + derived_a = data[["symbol_id", "date"]].copy() + derived_a["custom_value"] = 1.0 + derived_b = data[["symbol_id", "date"]].copy() + derived_b["custom_value"] = 2.0 + + with pytest.raises(ValueError, match="conflict"): + join_feature_frames(data, [derived_a, derived_b]) + + close_collision = data[["symbol_id", "date"]].copy() + close_collision["close"] = 1.0 + with pytest.raises(ValueError, match="conflict"): + join_feature_frames(data, [close_collision]) + + +def test_legacy_feature_cli_delegates_to_derived_registry(tmp_path): + runner = CliRunner() + minute_path = tmp_path / "minute.pq" + out_dir = tmp_path / "features" + _minute_bars().to_parquet(minute_path, index=False) + + list_result = runner.invoke(cli, ["feature", "list"]) + assert list_result.exit_code == 0, list_result.output + assert "minute_daily_summary" in list_result.output + + compute_result = runner.invoke(cli, [ + "feature", "compute", + "--minute-path", str(minute_path), + "--feature-type", "minute_daily_summary", + "--feature-name", "minute_summary", + "--output-dir", str(out_dir), + ]) + assert compute_result.exit_code == 0, compute_result.output + assert (out_dir / "minute_summary.pq").exists() + diff --git a/tests/test_features.py b/tests/test_features.py index dc0ad55..f572b2d 100644 --- a/tests/test_features.py +++ b/tests/test_features.py @@ -7,6 +7,7 @@ import pandas as pd import pytest from pipeline.features.compute import compute_feature, validate_feature_frame +from pipeline.features.library.minute_daily_summary import MinuteDailySummaryFeature from pipeline.features.registry import ( available_features, get_feature, @@ -68,6 +69,14 @@ def test_built_in_minute_daily_summary(): assert pd.isna(missing["minute_vwap"]) +def test_minute_daily_summary_feature_preserves_legacy_positional_compute(): + direct = MinuteDailySummaryFeature().compute(_minute_bars()) + via_registry = get_feature("minute_daily_summary").compute(_minute_bars()) + + assert "minute_vwap" in direct.columns + pd.testing.assert_frame_equal(direct, via_registry) + + def test_load_external_feature_module_and_filter_params(tmp_path): module_path = tmp_path / "external_feature.py" module_path.write_text(textwrap.dedent('''