From 94ab679a75595e30f94d89aac9e2c4279b005777 Mon Sep 17 00:00:00 2001 From: Yuxuan Yan Date: Wed, 10 Jun 2026 11:23:04 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20add=20portfolio=20phase=20=E2=80=94=20d?= =?UTF-8?q?iscretize=20alpha=20weights=20into=20tradable=20positions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a fourth pipeline phase modeling A-share microstructure: lot sizes, the 2023-08-10 Main Board increment change, STAR 200-share minimum/odd-lot rules, limit-up/down, suspensions, volume caps, costs, and slippage. Two layers: research (continuous weights → return/Sharpe/turnover/Fitness, no IC per repo convention) and execution (state-dependent lot rounding + two-stage greedy exposure repair + next-open reference simulator). Wires `portfolio build/simulate/eval` into the CLI and adds the POSITION/FILL/PNL schema contracts. Covered by tests/test_portfolio.py. Co-Authored-By: Claude Opus 4.7 --- .gitignore | 3 + cli.py | 9 +- pipeline/common/schema.py | 40 +++ pipeline/portfolio/__init__.py | 6 + pipeline/portfolio/cli.py | 100 ++++++++ pipeline/portfolio/constraints.py | 141 +++++++++++ pipeline/portfolio/construct.py | 171 +++++++++++++ pipeline/portfolio/discretize.py | 294 +++++++++++++++++++++ pipeline/portfolio/market_rules.py | 235 +++++++++++++++++ pipeline/portfolio/research.py | 89 +++++++ pipeline/portfolio/simulator.py | 255 +++++++++++++++++++ tests/test_portfolio.py | 394 +++++++++++++++++++++++++++++ 12 files changed, 1734 insertions(+), 3 deletions(-) create mode 100644 pipeline/portfolio/__init__.py create mode 100644 pipeline/portfolio/cli.py create mode 100644 pipeline/portfolio/constraints.py create mode 100644 pipeline/portfolio/construct.py create mode 100644 pipeline/portfolio/discretize.py create mode 100644 pipeline/portfolio/market_rules.py create mode 100644 pipeline/portfolio/research.py create mode 100644 pipeline/portfolio/simulator.py create mode 100644 tests/test_portfolio.py diff --git a/.gitignore b/.gitignore index 10c5dd1..bd8ae0f 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,6 @@ data/daily_bars/ alphas/ combos/ reports/ +/portfolio/ +/fills/ +/pnl/ diff --git a/cli.py b/cli.py index 0eae281..678a7e8 100644 --- a/cli.py +++ b/cli.py @@ -2,9 +2,10 @@ """Chinese Equity Quant Pipeline — decoupled phase CLI. Phases: - data — Download daily bars to parquet - alpha — Compute alpha weights from data - combo — Combine alphas into a single weight + data — Download daily bars to parquet + alpha — Compute alpha weights from data + combo — Combine alphas into a single weight + portfolio — Build tradable positions and simulate execution """ import logging @@ -14,6 +15,7 @@ import click from pipeline.data.cli import data from pipeline.alpha.cli import alpha from pipeline.combo.cli import combo +from pipeline.portfolio.cli import portfolio from tools.pqcat import pqcat from tools.alphaview import alphaview @@ -39,6 +41,7 @@ def cli(log_level): cli.add_command(data) cli.add_command(alpha) cli.add_command(combo) +cli.add_command(portfolio) cli.add_command(pqcat) cli.add_command(alphaview) diff --git a/pipeline/common/schema.py b/pipeline/common/schema.py index 85bea8d..da27ce8 100644 --- a/pipeline/common/schema.py +++ b/pipeline/common/schema.py @@ -42,3 +42,43 @@ COMBO_COLUMNS: Final[list[str]] = [ "combo_name", # str: identifies which combo (e.g. 'equal_weight') "weight", # float64: combined weight, signed ] + +# Required columns for portfolio (position) parquet files. +# A position is a tradable integer holding derived from a target weight under +# A-share lot/board rules. Produced by the `portfolio build` phase. +POSITION_COLUMNS: Final[list[str]] = [ + "symbol_id", # str + "date", # date + "portfolio_name", # str: identifies this construction run + "target_weight", # float64: w = alpha / sum(|alpha|); signed, sum(|w|)=1 + "target_value", # float64: v_target = booksize * w (signed dollar exposure) + "target_shares", # float64: q_target = v_target / price (continuous, signed) + "position_shares", # int64: discretized + repaired integer shares (signed) + "position_value", # float64: position_shares * price (signed) + "price", # float64: construction price (close by default) +] + +# Required columns for execution-simulator fill parquet files. +FILL_COLUMNS: Final[list[str]] = [ + "symbol_id", # str + "date", # date: the EXECUTION date (open[t+1] of the target date) + "portfolio_name", # str + "prev_shares", # int64: realized position carried in + "target_shares", # int64: requested target for this execution + "traded_shares", # int64: signed delta actually executed + "realized_shares", # int64: resulting position (blocked trades revert to prev) + "blocked", # int: 1 if the trade was (fully or partially) blocked + "trade_cost", # float64: commission + slippage in yuan +] + +# Required columns for execution-simulator per-day PnL parquet files. +PNL_COLUMNS: Final[list[str]] = [ + "date", # date + "portfolio_name", # str + "gross_exposure", # float64: sum(|position_value|) + "net_exposure", # float64: sum(signed position_value) + "pnl", # float64: daily mark-to-market PnL (yuan), net of cost + "cost", # float64: total trade cost that day (yuan) + "turnover", # float64: sum(|traded_value|) / booksize + "n_positions", # int: count of nonzero holdings +] diff --git a/pipeline/portfolio/__init__.py b/pipeline/portfolio/__init__.py new file mode 100644 index 0000000..472ad0a --- /dev/null +++ b/pipeline/portfolio/__init__.py @@ -0,0 +1,6 @@ +"""Portfolio construction phase. + +Turns continuous alpha/combo weights into tradable A-share positions under +date-aware lot/board rules, and simulates execution with market constraints, +costs, and slippage. +""" diff --git a/pipeline/portfolio/cli.py b/pipeline/portfolio/cli.py new file mode 100644 index 0000000..f9fc9cd --- /dev/null +++ b/pipeline/portfolio/cli.py @@ -0,0 +1,100 @@ +"""CLI for the portfolio construction and execution-simulation phase.""" + +import os + +import click +import pandas as pd + +from pipeline.portfolio.constraints import available_constraints, get_constraint +from pipeline.portfolio.construct import construct_positions +from pipeline.portfolio.research import evaluate_portfolio +from pipeline.portfolio.simulator import ReferenceSimulator + + +@click.group(name="portfolio") +def portfolio(): + """Construct tradable positions from weights and simulate execution.""" + + +@portfolio.command("build") +@click.option("--weights-path", required=True, help="Alpha or combo parquet (signed weights)") +@click.option("--data-path", required=True, help="Data parquet file or dataset directory") +@click.option("--booksize", type=float, required=True, help="Gross dollar exposure B") +@click.option("--portfolio-name", required=True, help="Name for this portfolio run") +@click.option("--price-field", default="close", help="Data column used as construction price") +@click.option("--output-dir", default="portfolio", help="Directory to save the positions parquet") +def build(weights_path, data_path, booksize, portfolio_name, price_field, output_dir): + """Discretize target weights into a tradable integer position book.""" + weights = pd.read_parquet(weights_path) + data = pd.read_parquet(data_path) + + result = construct_positions( + weights_df=weights, data_df=data, booksize=booksize, + portfolio_name=portfolio_name, price_field=price_field, + ) + + os.makedirs(output_dir, exist_ok=True) + out_path = f"{output_dir}/{portfolio_name}.pq" + result.to_parquet(out_path, index=False) + click.echo(f"Saved positions: {out_path} ({len(result):,} rows)") + per_date = result.groupby("date").agg( + gross=("position_value", lambda s: s.abs().sum()), + net=("position_value", "sum"), + ) + click.echo( + f"Gross exposure — mean: {per_date['gross'].mean():,.0f} " + f"(target {booksize:,.0f}); |net| mean: {per_date['net'].abs().mean():,.0f}" + ) + + +@portfolio.command("simulate") +@click.option("--positions-path", required=True, help="Positions parquet from `portfolio build`") +@click.option("--data-path", required=True, help="Data parquet file or dataset directory") +@click.option("--constraint", "constraints", multiple=True, + help=f"Trade constraint to apply (repeatable). Options: {available_constraints()}") +@click.option("--cost-bps", type=float, default=0.0, help="Commission in basis points") +@click.option("--slippage-bps", type=float, default=0.0, help="Slippage in basis points") +@click.option("--volume-frac", type=float, default=0.10, + help="Max traded value as a fraction of daily turnover (volume_cap)") +@click.option("--output-dir", default=".", help="Base dir; writes fills/ and pnl/ subdirs") +def simulate(positions_path, data_path, constraints, cost_bps, slippage_bps, + volume_frac, output_dir): + """Simulate next-open execution under A-share constraints, costs, slippage.""" + positions = pd.read_parquet(positions_path) + data = pd.read_parquet(data_path) + name = positions["portfolio_name"].iloc[0] if len(positions) else "portfolio" + + built = [] + for c in constraints: + params = {"max_frac": volume_frac} if c == "volume_cap" else {} + built.append(get_constraint(c, **params)) + + sim = ReferenceSimulator(constraints=built, cost_bps=cost_bps, slippage_bps=slippage_bps) + fills, pnl = sim.run(positions, data) + + fills_dir = os.path.join(output_dir, "fills") + pnl_dir = os.path.join(output_dir, "pnl") + os.makedirs(fills_dir, exist_ok=True) + os.makedirs(pnl_dir, exist_ok=True) + fills.to_parquet(f"{fills_dir}/{name}.pq", index=False) + pnl.to_parquet(f"{pnl_dir}/{name}.pq", index=False) + click.echo(f"Saved fills: {fills_dir}/{name}.pq ({len(fills):,} rows)") + click.echo(f"Saved pnl: {pnl_dir}/{name}.pq ({len(pnl):,} rows)") + if len(pnl): + click.echo( + f"Total PnL: {pnl['pnl'].sum():,.0f} | total cost: {pnl['cost'].sum():,.0f} " + f"| blocked trades: {int(fills['blocked'].sum()):,}" + ) + + +@portfolio.command("eval") +@click.option("--positions-path", required=True, help="Positions parquet from `portfolio build`") +@click.option("--data-path", required=True, help="Data parquet file or dataset directory") +def eval_(positions_path, data_path): + """Print Layer-1 research metrics (return/Sharpe/turnover/max-dd/Fitness; no IC).""" + positions = pd.read_parquet(positions_path) + data = pd.read_parquet(data_path) + metrics = evaluate_portfolio(positions, data) + click.echo("Research-portfolio metrics:") + for key, value in metrics.items(): + click.echo(f" {key:18s}: {value}") diff --git a/pipeline/portfolio/constraints.py b/pipeline/portfolio/constraints.py new file mode 100644 index 0000000..5f91da4 --- /dev/null +++ b/pipeline/portfolio/constraints.py @@ -0,0 +1,141 @@ +"""Trade constraints for the execution simulator. + +A constraint answers: *given today's market state, how much may each name be +traded?* Each returns per-name signed delta bounds ``(low, high)`` in shares; +the simulator intersects them (``low = max(lows)``, ``high = min(highs)``) and +clips the desired trade. ``-inf/inf`` mean uncapped, ``0`` blocks that direction. + +Constraints self-register via :func:`register_constraint` (mirroring the alpha +registry) so the CLI can select them by name. They consume the abstracted +:class:`~pipeline.portfolio.market_rules.LimitStatus` rather than raw prices, +leaving room for richer fill models (一字板, queues, partial fills) later. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Type + +import numpy as np + +from pipeline.portfolio.market_rules import LimitStatus + +if TYPE_CHECKING: + from pipeline.portfolio.simulator import TradeContext + + +class TradeConstraint(ABC): + """Per-name tradeability rule producing signed share-delta bounds.""" + + name: str = "" + + @abstractmethod + def delta_bounds(self, ctx: "TradeContext") -> tuple[np.ndarray, np.ndarray]: + """Return ``(low, high)`` arrays: the min/max signed share delta allowed.""" + + def adjust_targets(self, ctx: "TradeContext") -> np.ndarray | None: + """Optional portfolio-level retargeting hook (e.g. neutrality). + + Returns a new target-shares array, or None to leave targets unchanged. + Default: no adjustment. Future industry/beta neutrality can implement a + cheap numpy least-squares projection here — no MIP needed. + """ + return None + + +_CONSTRAINTS: dict[str, Type[TradeConstraint]] = {} + + +def register_constraint(cls: Type[TradeConstraint]) -> Type[TradeConstraint]: + """Class decorator registering a constraint under its ``name``.""" + if not (isinstance(cls, type) and issubclass(cls, TradeConstraint)): + raise TypeError(f"{cls!r} is not a TradeConstraint subclass") + key = getattr(cls, "name", "") + if not key: + raise ValueError(f"{cls.__name__} must set a non-empty class attribute `name`") + existing = _CONSTRAINTS.get(key) + if existing is not None and existing is not cls: + raise ValueError(f"Constraint name '{key}' already registered by {existing.__name__}") + _CONSTRAINTS[key] = cls + return cls + + +def available_constraints() -> list[str]: + """Sorted names of all registered constraints.""" + return sorted(_CONSTRAINTS) + + +def get_constraint(name: str, **params) -> TradeConstraint: + """Instantiate a registered constraint by name. + + Raises: + KeyError: If ``name`` is not registered. + """ + if name not in _CONSTRAINTS: + raise KeyError(f"Unknown constraint '{name}'. Available: {sorted(_CONSTRAINTS)}") + return _CONSTRAINTS[name](**params) + + +def _unbounded(n: int) -> tuple[np.ndarray, np.ndarray]: + return np.full(n, -np.inf), np.full(n, np.inf) + + +@register_constraint +class SuspensionConstraint(TradeConstraint): + """Suspended names (``tradestatus == 0``) cannot trade at all.""" + + name = "suspension" + + def delta_bounds(self, ctx): + n = len(ctx.target_shares) + low, high = _unbounded(n) + suspended = ctx.slice.tradestatus == 0 + low = np.where(suspended, 0.0, low) + high = np.where(suspended, 0.0, high) + return low, high + + +@register_constraint +class PriceLimitConstraint(TradeConstraint): + """Limit-up blocks buys; limit-down blocks sells. + + Consumes ``ctx.slice.limit_status`` (NORMAL/UP_LIMIT/DOWN_LIMIT), not raw + prices, so future states (e.g. limit-locked 一字板 with zero fill) plug in + by extending the status set. + """ + + name = "price_limit" + + def delta_bounds(self, ctx): + n = len(ctx.target_shares) + low, high = _unbounded(n) + status = ctx.slice.limit_status + at_up = status == LimitStatus.UP_LIMIT.value + at_down = status == LimitStatus.DOWN_LIMIT.value + high = np.where(at_up, 0.0, high) # cannot buy at the up limit + low = np.where(at_down, 0.0, low) # cannot sell at the down limit + return low, high + + +@register_constraint +class VolumeCapConstraint(TradeConstraint): + """Cap traded **value** at a fraction of the day's turnover value. + + ``|trade_value| ≤ max_frac · amount`` (amount = daily turnover in yuan), + converted to a share cap via the execution price. Value-based, not + share-count based. + """ + + name = "volume_cap" + + def __init__(self, max_frac: float = 0.10): + self.max_frac = max_frac + + def delta_bounds(self, ctx): + amount = np.asarray(ctx.slice.amount, dtype=np.float64) + price = np.asarray(ctx.slice.price, dtype=np.float64) + cap_value = self.max_frac * np.where(np.isfinite(amount), amount, 0.0) + with np.errstate(divide="ignore", invalid="ignore"): + cap_shares = np.where(price > 0, cap_value / price, 0.0) + cap_shares = np.floor(cap_shares) + return -cap_shares, cap_shares diff --git a/pipeline/portfolio/construct.py b/pipeline/portfolio/construct.py new file mode 100644 index 0000000..c4acae6 --- /dev/null +++ b/pipeline/portfolio/construct.py @@ -0,0 +1,171 @@ +"""Continuous portfolio construction and the date-ordered position driver. + +Layer-1 (research) math lives in :func:`continuous_targets`: it turns a signed +weight vector into target weights, dollar exposures, and continuous shares. +:func:`construct_positions` is the Layer-2 driver — it threads ``prev_shares`` +across dates (positions are stateful, unlike alphas/combos), discretizing and +repairing each day's target into a tradable integer book. + +Return-convention note: weights here are *target allocations*. The research +evaluation in :mod:`pipeline.portfolio.research` marks them close-to-close on the +*next* period (no look-ahead); the execution simulator marks the actually-filled +book at the next open. See those modules for details. +""" + +from __future__ import annotations + +import logging + +import numpy as np +import pandas as pd + +from pipeline.common.schema import POSITION_COLUMNS +from pipeline.portfolio.discretize import repair_exposure, round_to_valid_lot +from pipeline.portfolio.market_rules import MarketRule + +logger = logging.getLogger(__name__) + + +def continuous_targets( + alpha: np.ndarray, price: np.ndarray, booksize: float +) -> tuple[np.ndarray, np.ndarray, np.ndarray]: + """Continuous (research) portfolio targets from a signed weight vector. + + ``w = alpha / sum(|alpha|)`` so ``sum(|w|) = 1`` and, because the upstream + alpha is demeaned, ``sum(w) ≈ 0`` (dollar-neutral). Then + ``v_target = booksize · w`` and ``q_target = v_target / price``. + + NaN alphas and non-positive / NaN prices are treated as a 0 target. + + Args: + alpha: Signed weight vector, length N. + price: Per-name price (yuan), length N. + booksize: Gross dollar exposure ``B``. + + Returns: + ``(w, v_target, q_target)``, each a float array of length N. + """ + alpha = np.asarray(alpha, dtype=np.float64) + price = np.asarray(price, dtype=np.float64) + a = np.where(np.isfinite(alpha), alpha, 0.0) + gross = np.abs(a).sum() + if gross <= 0: + zeros = np.zeros_like(a) + return zeros, zeros.copy(), zeros.copy() + w = a / gross + v_target = booksize * w + tradable = np.isfinite(price) & (price > 0) + q_target = np.where(tradable, v_target / np.where(tradable, price, 1.0), 0.0) + # Names without a tradable price get no target exposure. + w = np.where(tradable, w, 0.0) + v_target = np.where(tradable, v_target, 0.0) + return w, v_target, q_target + + +def _pivot(df: pd.DataFrame, value: str, weight_col: str | None = None) -> pd.DataFrame: + col = weight_col or value + return df.pivot_table( + index="date", columns="symbol_id", values=col, aggfunc="first" + ).sort_index() + + +def construct_positions( + weights_df: pd.DataFrame, + data_df: pd.DataFrame, + booksize: float, + portfolio_name: str, + rule_engine: MarketRule | None = None, + price_field: str = "close", + net_tol: float = 0.02, + gross_tol: float = 0.02, +) -> pd.DataFrame: + """Build a tradable position book from target weights, day by day. + + Pivots weights and prices to a wide grid on a fixed symbol column order, + then iterates dates in ascending order carrying an integer ``prev_shares`` + vector. Each date: continuous targets → state-dependent lot rounding → + two-stage exposure repair. Names absent on a date get weight 0 (which closes + any stale holding). An empty / zero-gross cross-section carries the book + unchanged. + + Args: + weights_df: Long frame with ``symbol_id, date, weight`` (ALPHA/COMBO). + data_df: Long frame with DATA_COLUMNS (prices, tradestatus, isST). + booksize: Gross dollar exposure ``B``. + portfolio_name: Identifier stored in the ``portfolio_name`` column. + rule_engine: A :class:`MarketRule`; a default one is built if None. + price_field: Data column used as the construction price (default close). + net_tol: Net tolerance (fraction of B) passed to the repair. + gross_tol: Gross tolerance (fraction of B) passed to the repair. + + Returns: + Long DataFrame with POSITION_COLUMNS, sorted by ``(symbol_id, date)``. + """ + rule_engine = rule_engine or MarketRule() + + price_wide = _pivot(data_df, price_field) + w_wide = _pivot(weights_df, "weight") + st_wide = _pivot(data_df, "isST") if "isST" in data_df.columns else None + + # Fixed, sorted symbol-column order shared across the whole run. + symbols = sorted(set(price_wide.columns) | set(w_wide.columns)) + price_wide = price_wide.reindex(columns=symbols) + w_wide = w_wide.reindex(columns=symbols) + if st_wide is not None: + st_wide = st_wide.reindex(columns=symbols) + + dates = sorted(set(price_wide.index) & set(w_wide.index)) + if not dates: + logger.warning("No overlapping dates between weights and data; empty result.") + return pd.DataFrame(columns=POSITION_COLUMNS) + + sym_arr = np.asarray(symbols, dtype=object) + n = len(symbols) + prev_shares = np.zeros(n, dtype=np.int64) + + blocks: list[pd.DataFrame] = [] + for d in dates: + price = price_wide.loc[d].to_numpy(dtype=np.float64) + alpha = w_wide.loc[d].to_numpy(dtype=np.float64) + is_st = ( + st_wide.loc[d].fillna(0).to_numpy() if st_wide is not None + else np.zeros(n) + ) + min_open, increment, odd_full, _ = rule_engine.get_rules_vectorized( + sym_arr, d, is_st + ) + + w, v_target, q_target = continuous_targets(alpha, price, booksize) + q_round = round_to_valid_lot(q_target, prev_shares, min_open, increment, odd_full) + pos = repair_exposure( + q_round, q_target, price, increment, min_open, prev_shares, odd_full, + booksize=booksize, net_tol=net_tol, gross_tol=gross_tol, + ) + + safe_price = np.where(np.isfinite(price), price, 0.0) + blocks.append(pd.DataFrame({ + "symbol_id": symbols, + "date": d, + "portfolio_name": portfolio_name, + "target_weight": w, + "target_value": v_target, + "target_shares": q_target, + "position_shares": pos, + "position_value": pos.astype(np.float64) * safe_price, + "price": price, + })) + prev_shares = pos + + result = pd.concat(blocks, ignore_index=True) + # Drop names that are flat AND have no target (keep the active universe tidy). + active = (result["position_shares"] != 0) | (result["target_weight"] != 0) + result = result[active] + result = result[POSITION_COLUMNS] + result = result.sort_values(["symbol_id", "date"]).reset_index(drop=True) + + n_dates = result["date"].nunique() + logger.info( + "Portfolio '%s': %d symbols × %d dates, booksize %.0f", + portfolio_name, result["symbol_id"].nunique(), n_dates, booksize, + ) + return result diff --git a/pipeline/portfolio/discretize.py b/pipeline/portfolio/discretize.py new file mode 100644 index 0000000..63c8e7f --- /dev/null +++ b/pipeline/portfolio/discretize.py @@ -0,0 +1,294 @@ +"""Continuous → tradable position discretization and exposure repair. + +Pure-numpy, no I/O. Two steps: + +1. :func:`round_to_valid_lot` — snap continuous target shares to the nearest + *valid resting position* given the per-name lot rule AND the current holding + (``prev_shares``). Rounding is state-dependent: a target below the board + minimum cannot *open* a fresh lot, but an existing holding may always be + reduced to 0, and a 科创 odd-lot residual may be sold whole. + +2. :func:`repair_exposure` — a two-stage greedy that drives net exposure to ~0 + (Stage A) and gross exposure to the booksize (Stage B) while minimizing the + dollar-space tracking error ``sum((v_i - v_target_i)**2)``. Splitting the two + stages avoids the oscillation a single mixed loop suffers (gross repair + breaking net neutrality and vice versa). O(N log N) via lazy heaps. + +A "move" adjusts one name by ±``increment`` shares (or closes it to 0 at the +lattice boundary); names are never opened during repair and never flip sign. +""" + +from __future__ import annotations + +import heapq +import itertools + +import numpy as np + + +def round_to_valid_lot( + target: np.ndarray, + prev_shares: np.ndarray, + min_open: np.ndarray, + increment: np.ndarray, + sell_is_odd_full: np.ndarray | None = None, +) -> np.ndarray: + """Snap continuous target shares to valid integer resting positions. + + The valid resting lattice for a name is ``{0} ∪ {min_open + k·increment : + k ≥ 0}`` on each side. Rounding depends on the current holding: + + * **Opening** (no holding on the target side) a magnitude below ``min_open`` + is not allowed → snaps to 0. + * **Holding** the same side, a sub-minimum target snaps to the nearer of + ``{0, min_open}`` (so a reduction can rest at the minimum lot or close out). + * A full liquidation to 0 is always valid (covers the 科创 odd-lot sell: + a residual ``< min_open`` can only be sold whole, i.e. to 0). + * Sign is never flipped unless ``target`` itself flipped sign. + + Args: + target: Continuous signed target shares, length N. + prev_shares: Current signed integer holding, length N. + min_open: Per-name minimum open size, length N. + increment: Per-name share increment (> 0), length N. + sell_is_odd_full: Unused for resting validity (odd-lot sells already + resolve to 0); accepted for API symmetry and documentation. + + Returns: + ``int64`` array of valid resting positions, length N. + """ + target = np.asarray(target, dtype=np.float64) + prev = np.asarray(prev_shares, dtype=np.int64) + min_open = np.asarray(min_open, dtype=np.float64) + increment = np.asarray(increment, dtype=np.float64) + + sign = np.sign(target) + mag = np.abs(target) + + # Lattice magnitude for mag >= min_open: min_open + round((mag-min_open)/inc)*inc + k = np.maximum(np.round((mag - min_open) / increment), 0.0) + lattice_mag = min_open + k * increment + + holding_same_side = (prev != 0) & (np.sign(prev) == sign) & (sign != 0) + + # Sub-minimum handling: opening -> 0; holding same side -> nearer of {0, min_open}. + sub_min = mag < min_open + sub_min_mag = np.where( + holding_same_side & (mag >= 0.5 * min_open), min_open, 0.0 + ) + final_mag = np.where(sub_min, sub_min_mag, lattice_mag) + rounded = sign * final_mag + return rounded.astype(np.int64) + + +def _exposures(q: np.ndarray, price: np.ndarray) -> tuple[np.ndarray, float, float]: + v = q.astype(np.float64) * price + return v, float(v.sum()), float(np.abs(v).sum()) + + +def repair_exposure( + q_round: np.ndarray, + q_target: np.ndarray, + price: np.ndarray, + increment: np.ndarray, + min_open: np.ndarray, + prev_shares: np.ndarray, + sell_is_odd_full: np.ndarray | None = None, + booksize: float = 1.0, + net_tol: float = 0.02, + gross_tol: float = 0.02, + max_iters: int | None = None, +) -> np.ndarray: + """Two-stage greedy exposure repair in dollar space. + + Stage A drives ``net = sum(v_i)`` toward 0; Stage B drives ``gross = + sum(|v_i|)`` toward ``booksize`` using only moves that keep ``|net|`` within + its tolerance band, so Stage B cannot undo Stage A. Both stages pick, at each + step, the admissible ±``increment`` move with the lowest tracking-error cost + per dollar moved (``ΔTE/|Δv|`` where ``ΔTE = 2·Δv·(v_i - v_target_i) + + Δv²``). Names that round to 0 are never re-opened here. + + Tolerances are fractions of ``booksize`` but floored to the lot granularity: + with coarse lots (e.g. pre-2023 100-share main-board lots) exact neutrality + is unreachable, so the floor prevents a deadlock / infinite loop. + + Args: + q_round: Integer positions from :func:`round_to_valid_lot`, length N. + q_target: Continuous target shares (the tracking anchor), length N. + price: Per-name price (yuan), length N. + increment: Per-name share increment (> 0), length N. + min_open: Per-name minimum open size, length N. + prev_shares: Current holding (unused directly; reserved for borrow caps). + sell_is_odd_full: Reserved; accepted for API symmetry. + booksize: Target gross exposure ``B``. + net_tol: Net tolerance as a fraction of ``B``. + gross_tol: Gross tolerance as a fraction of ``B``. + max_iters: Hard cap on repair moves (default ``8·N``). + + Returns: + ``int64`` repaired positions, length N. + """ + q = np.asarray(q_round, dtype=np.int64).copy() + price = np.asarray(price, dtype=np.float64) + increment = np.asarray(increment, dtype=np.int64).astype(np.float64) + min_open = np.asarray(min_open, dtype=np.int64).astype(np.float64) + qt = np.asarray(q_target, dtype=np.float64) + n = len(q) + if n == 0: + return q + + vt = np.where(np.isfinite(qt), qt, 0.0) * price # v_target, NaN-safe + tradable = np.isfinite(price) & (price > 0) + step = np.where(tradable, increment * price, np.inf) # dollar per increment + + if max_iters is None: + max_iters = 8 * n + + # Adaptive absolute tolerances: never finer than the lot granularity. + active_step = step[(q != 0) & tradable] + max_step = float(active_step.max()) if active_step.size else 0.0 + min_step = float(active_step.min()) if active_step.size else 0.0 + net_tol_abs = max(net_tol * booksize, max_step) + gross_tol_abs = max(gross_tol * booksize, min_step) + net_band = net_tol_abs # Stage B keeps |net| within this band + + v, net, gross = _exposures(q, price) + + def _move(i: int, grow: bool): + """Return (dshares, dv, dte) for a grow/shrink move on name i, or None.""" + if q[i] == 0 or not tradable[i]: + return None + s = 1 if q[i] > 0 else -1 + if grow: + dshares = s * int(increment[i]) + else: + mag = abs(int(q[i])) + if mag - increment[i] >= min_open[i]: + dshares = -s * int(increment[i]) + else: + dshares = -int(q[i]) # close to 0 (lattice boundary / odd lot) + if dshares == 0: + return None + dv = dshares * price[i] + dte = 2.0 * dv * (v[i] - vt[i]) + dv * dv + return dshares, dv, dte + + def _apply(i: int, dshares: int, dv: float): + nonlocal net, gross + old_abs = abs(v[i]) + q[i] += dshares + v[i] += dv + net += dv + gross += abs(v[i]) - old_abs + + counter = itertools.count() + active_idx = np.nonzero((q != 0) & tradable)[0] + + # ---- Stage A: net repair ------------------------------------------------- + def _stageA_dir() -> int: + return -1 if net > 0 else 1 # desired sign of dv + + iters = 0 + while abs(net) > net_tol_abs and iters < max_iters: + want = _stageA_dir() # dv sign we need + heap: list = [] + best_key: dict[int, float] = {} + for i in active_idx: + i = int(i) + # For net>0 (want dv<0): shrink longs, grow shorts. Mirror otherwise. + grow = (q[i] < 0) if want < 0 else (q[i] > 0) + mv = _move(i, grow) + if mv is None: + continue + _, dv, dte = mv + if np.sign(dv) != want: + continue + key = dte / abs(dv) + best_key[i] = key + heapq.heappush(heap, (key, next(counter), i, grow)) + if not heap: + break + progressed = False + while heap and abs(net) > net_tol_abs and iters < max_iters: + key, _, i, grow = heapq.heappop(heap) + if best_key.get(i) != key: + continue # stale + mv = _move(i, grow) + if mv is None: + best_key.pop(i, None) + continue + dshares, dv, dte = mv + if np.sign(dv) != want: + best_key.pop(i, None) + continue + # Don't overshoot net through 0 by more than the tolerance band. + if abs(net + dv) > abs(net) and abs(net + dv) > net_tol_abs: + best_key.pop(i, None) + continue + _apply(i, dshares, dv) + iters += 1 + progressed = True + if q[i] != 0: + nk = _move(i, grow) + if nk is not None: + _, ndv, ndte = nk + if np.sign(ndv) == want: + k2 = ndte / abs(ndv) + best_key[i] = k2 + heapq.heappush(heap, (k2, next(counter), i, grow)) + continue + best_key.pop(i, None) + if not progressed: + break + + # ---- Stage B: gross repair (net-preserving) ----------------------------- + iters = 0 + active_idx = np.nonzero((q != 0) & tradable)[0] + while abs(gross - booksize) > gross_tol_abs and iters < max_iters: + grow = gross < booksize # need more gross → grow magnitudes; else shrink + heap = [] + best_key = {} + for i in active_idx: + i = int(i) + mv = _move(i, grow) + if mv is None: + continue + _, dv, dte = mv + # Net-band filter: never push |net| past the band. + if abs(net + dv) > net_band and abs(net + dv) >= abs(net): + continue + key = dte / abs(dv) + best_key[i] = key + heapq.heappush(heap, (key, next(counter), i, grow)) + if not heap: + break + progressed = False + while heap and abs(gross - booksize) > gross_tol_abs and iters < max_iters: + key, _, i, g = heapq.heappop(heap) + if best_key.get(i) != key: + continue + mv = _move(i, g) + if mv is None: + best_key.pop(i, None) + continue + dshares, dv, dte = mv + if abs(net + dv) > net_band and abs(net + dv) >= abs(net): + best_key.pop(i, None) + continue + _apply(i, dshares, dv) + iters += 1 + progressed = True + if q[i] != 0: + nk = _move(i, g) + if nk is not None: + _, ndv, ndte = nk + if not (abs(net + ndv) > net_band and abs(net + ndv) >= abs(net)): + k2 = ndte / abs(ndv) + best_key[i] = k2 + heapq.heappush(heap, (k2, next(counter), i, g)) + continue + best_key.pop(i, None) + if not progressed: + break + + return q diff --git a/pipeline/portfolio/market_rules.py b/pipeline/portfolio/market_rules.py new file mode 100644 index 0000000..d4b5249 --- /dev/null +++ b/pipeline/portfolio/market_rules.py @@ -0,0 +1,235 @@ +"""Date-aware A-share market rule engine. + +The engine is deliberately separated from alpha/portfolio logic: it answers a +single question — *what lot, increment, sell, and price-limit rules apply to a +given symbol on a given date* — from a data-driven table. New rule changes or +new boards are added by appending rows to :data:`RULE_TABLE`; no branching logic +needs editing. + +Boards are detected from the internal ``symbol_id`` prefix (``sh600000`` / +``sz000001`` / ``sh688981`` / ``sz300750``). +""" + +from __future__ import annotations + +import datetime as _dt +from dataclasses import dataclass +from enum import Enum + +import numpy as np + + +class Board(str, Enum): + """A-share trading board.""" + + MAIN = "main" # sh60xxxx, sz000/001/002xxx (沪深主板, incl. former SME) + STAR = "star" # sh688xxx (科创板 / STAR Market) + CHINEXT = "chinext" # sz300xxx (创业板 / ChiNext) + UNKNOWN = "unknown" + + +class LimitStatus(int, Enum): + """Daily price-limit state of a name on a given date. + + Constraints consume this status rather than comparing raw prices, so future + refinements (一字板 / limit-locked, queue priority, partial fills) only add + states or richer fill logic without rewriting constraints. + """ + + NORMAL = 0 + UP_LIMIT = 1 + DOWN_LIMIT = -1 + + +@dataclass(frozen=True) +class Rule: + """Lot/sell/price-limit rule that applies to one (board, date) cell.""" + + minimum_open_size: int # min shares to OPEN (buy) a position + share_increment: int # lot granularity above the minimum + sell_rule: str # "lot" | "odd_lot_full" (odd residual sellable whole) + price_limit_pct: float # daily up/down band as a fraction (e.g. 0.10) + + +@dataclass(frozen=True) +class RuleSpan: + """A rule that is valid for ``[valid_from, valid_to)`` on a board.""" + + board: Board + valid_from: _dt.date + valid_to: _dt.date + rule: Rule + + +# --- The rule table ---------------------------------------------------------- +# Append rows here for future rule changes or new boards. Order does not matter; +# get_rule selects by board + [valid_from, valid_to) membership. +_MAIN_INCREMENT_CHANGE = _dt.date(2023, 8, 10) +_DATE_MIN = _dt.date(1990, 1, 1) +_DATE_MAX = _dt.date(2999, 12, 31) + +#: Price-limit band applied to ST names, overriding the board band. +ST_PRICE_LIMIT_PCT = 0.05 + +RULE_TABLE: list[RuleSpan] = [ + # 沪深主板: before 2023-08-10 orders must be whole multiples of 100; + # on/after 2023-08-10 the minimum is still 100 but the increment is 1 share. + RuleSpan(Board.MAIN, _DATE_MIN, _MAIN_INCREMENT_CHANGE, + Rule(100, 100, "lot", 0.10)), + RuleSpan(Board.MAIN, _MAIN_INCREMENT_CHANGE, _DATE_MAX, + Rule(100, 1, "lot", 0.10)), + # 科创板 (STAR): from launch, min buy 200, increment 1; a residual holding + # below 200 (an odd lot) may be sold in full. + RuleSpan(Board.STAR, _DATE_MIN, _DATE_MAX, + Rule(200, 1, "odd_lot_full", 0.20)), + # 创业板 (ChiNext): APPROXIMATION — modeled as post-2023 main-board lots + # (min 100, increment 1) with a 20% band. Real ChiNext history (e.g. the + # 2020-08-24 registration-system 20% band, earlier 10% band, 100-share lots) + # can be added as extra rows here WITHOUT touching get_rule's logic. + RuleSpan(Board.CHINEXT, _DATE_MIN, _DATE_MAX, + Rule(100, 1, "lot", 0.20)), +] + +#: Fallback for UNKNOWN boards — conservative main-board-like lots, 10% band. +_DEFAULT_RULE = Rule(100, 100, "lot", 0.10) + + +def detect_board(symbol_id: str) -> Board: + """Classify a symbol into its trading board from the ``symbol_id`` prefix. + + Args: + symbol_id: Internal code like ``sh600000`` / ``sz300750``. + + Returns: + The :class:`Board`; :attr:`Board.UNKNOWN` if no rule matches. + """ + if len(symbol_id) < 5: + return Board.UNKNOWN + exchange, code = symbol_id[:2], symbol_id[2:] + if exchange == "sh": + if code.startswith("688"): + return Board.STAR + if code.startswith("60"): + return Board.MAIN + elif exchange == "sz": + if code.startswith("300"): + return Board.CHINEXT + if code[:3] in ("000", "001", "002"): + return Board.MAIN + return Board.UNKNOWN + + +def _to_date(value) -> _dt.date: + """Coerce a date / datetime / pandas Timestamp / ISO string to ``date``.""" + if isinstance(value, _dt.datetime): + return value.date() + if isinstance(value, _dt.date): + return value + # numpy datetime64 / pandas Timestamp / str all accept str() round-trip. + return _dt.date.fromisoformat(str(value)[:10]) + + +class MarketRule: + """Resolve lot/sell/price-limit rules for a symbol on a date. + + The engine is stateless; instantiate once and reuse across the run. + """ + + def __init__(self, table: list[RuleSpan] | None = None, + default_rule: Rule = _DEFAULT_RULE, + st_price_limit_pct: float = ST_PRICE_LIMIT_PCT) -> None: + self._table = table if table is not None else RULE_TABLE + self._default = default_rule + self._st_limit = st_price_limit_pct + + def get_rule(self, symbol_id: str, on, is_st: bool = False) -> Rule: + """Return the :class:`Rule` for ``symbol_id`` on date ``on``. + + Args: + symbol_id: Internal symbol code. + on: Trading date (``date``, ``datetime``, ``Timestamp``, or ISO str). + is_st: If True, override the price-limit band with the ST band. + + Returns: + The matching :class:`Rule`, with ST band applied if ``is_st``. + """ + day = _to_date(on) + board = detect_board(symbol_id) + rule = self._default + for span in self._table: + if span.board is board and span.valid_from <= day < span.valid_to: + rule = span.rule + break + if is_st: + rule = Rule(rule.minimum_open_size, rule.share_increment, + rule.sell_rule, self._st_limit) + return rule + + def get_rules_vectorized( + self, symbol_ids, on, is_st, + ) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: + """Vectorized rule lookup for a whole cross-section on one date. + + Args: + symbol_ids: Sequence of ``symbol_id`` strings (length N). + on: The trading date (shared by all names). + is_st: Boolean array (length N), 1/True where the name is ST. + + Returns: + Tuple of four numpy arrays, each length N: + ``(min_open, increment, sell_is_odd_full, limit_pct)`` with dtypes + ``int64, int64, bool, float64``. + """ + symbol_ids = np.asarray(symbol_ids, dtype=object) + is_st = np.asarray(is_st).astype(bool) + n = len(symbol_ids) + min_open = np.empty(n, dtype=np.int64) + increment = np.empty(n, dtype=np.int64) + odd_full = np.empty(n, dtype=bool) + limit_pct = np.empty(n, dtype=np.float64) + + # Resolve once per distinct symbol (board only depends on the prefix). + cache: dict[str, Rule] = {} + for i, sym in enumerate(symbol_ids): + rule = cache.get(sym) + if rule is None: + rule = self.get_rule(sym, on, is_st=False) + cache[sym] = rule + min_open[i] = rule.minimum_open_size + increment[i] = rule.share_increment + odd_full[i] = rule.sell_rule == "odd_lot_full" + limit_pct[i] = self._st_limit if is_st[i] else rule.price_limit_pct + return min_open, increment, odd_full, limit_pct + + +def compute_limit_status( + price, preclose, limit_pct, *, tol: float = 1e-6, +) -> np.ndarray: + """Classify each name's daily price-limit state. + + A name is at the up (down) limit when its price reaches + ``preclose * (1 ± limit_pct)`` within ``tol`` relative tolerance. + + Args: + price: Reference price array (e.g. the open at which we trade). + preclose: Previous close array. + limit_pct: Per-name daily band fraction. + tol: Relative tolerance for the limit comparison. + + Returns: + ``int8`` array of :class:`LimitStatus` values (length N). Names with + non-positive or NaN preclose are treated as ``NORMAL``. + """ + price = np.asarray(price, dtype=np.float64) + preclose = np.asarray(preclose, dtype=np.float64) + limit_pct = np.asarray(limit_pct, dtype=np.float64) + + status = np.zeros(price.shape, dtype=np.int8) + valid = np.isfinite(price) & np.isfinite(preclose) & (preclose > 0) + up = preclose * (1.0 + limit_pct) + down = preclose * (1.0 - limit_pct) + at_up = valid & (price >= up * (1.0 - tol)) + at_down = valid & (price <= down * (1.0 + tol)) + status[at_up] = LimitStatus.UP_LIMIT.value + status[at_down] = LimitStatus.DOWN_LIMIT.value + return status diff --git a/pipeline/portfolio/research.py b/pipeline/portfolio/research.py new file mode 100644 index 0000000..25ff995 --- /dev/null +++ b/pipeline/portfolio/research.py @@ -0,0 +1,89 @@ +"""Layer-1 (research) evaluation of a portfolio's target weights. + +This is the WorldQuant-style research view: continuous target weights, no lot or +trading constraints. Metrics are return / Sharpe / turnover / max-drawdown / +**Fitness**. There is deliberately **no IC/IR** — consistent with the repo's +convention that an alpha is a position weight, not a return predictor. + +Return convention (documented): the target weight formed from information at +date ``t`` earns the *next* period's close-to-close return, i.e. weights are +shifted one day relative to realized returns, so there is no look-ahead: +``R_t = sum_i w_{i,t} · r_{i,t+1}`` normalized by gross exposure. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +#: WorldQuant fitness floor on turnover (avoids dividing by ~0 turnover). +_TURNOVER_FLOOR = 0.125 + + +def evaluate_portfolio(positions_df: pd.DataFrame, data_df: pd.DataFrame) -> dict: + """Evaluate target weights as a continuous research portfolio. + + Args: + positions_df: POSITION_COLUMNS (uses ``target_weight``). + data_df: DATA_COLUMNS (uses ``close`` for returns). + + Returns: + Dict with ``cumulative_return, sharpe_annual, turnover_annual, + max_drawdown, fitness, hit_rate, n_dates``. No IC key. + """ + close = data_df.pivot_table( + index="date", columns="symbol_id", values="close", aggfunc="first" + ).sort_index() + returns = close.pct_change() + + weights = positions_df.pivot_table( + index="date", columns="symbol_id", values="target_weight", aggfunc="first" + ).sort_index() + + common = weights.index.intersection(returns.index) + weights = weights.loc[common] + returns = returns.loc[common] + + empty = { + "cumulative_return": 0.0, "sharpe_annual": 0.0, "turnover_annual": 0.0, + "max_drawdown": 0.0, "fitness": 0.0, "hit_rate": 0.0, + "n_dates": len(common), + } + if len(common) < 3: + return empty + + gross = weights.abs().sum(axis=1) + # Weights at t earn the return from t to t+1: shift returns back by one. + fwd = returns.shift(-1) + daily = (weights * fwd).sum(axis=1) / gross.replace(0.0, np.nan) + daily = daily.dropna() + if len(daily) < 2: + return empty + + cumulative_return = float((1.0 + daily).prod() - 1.0) + mu, sigma = daily.mean(), daily.std() + sharpe_annual = float(np.sqrt(252) * mu / sigma) if sigma > 0 else 0.0 + + weight_change = weights.diff().abs().sum(axis=1) + prev_gross = gross.shift(1) + daily_turnover = (weight_change / prev_gross.replace(0.0, np.nan)).dropna() + turnover_annual = float(daily_turnover.mean() * 252) + + equity = (1.0 + daily).cumprod() + drawdown = (equity - equity.cummax()) / equity.cummax() + max_drawdown = float(drawdown.min()) + + # Fitness = Sharpe · sqrt(|annualized return| / max(annual turnover, floor)). + ann_return = float(mu * 252) + denom = max(turnover_annual, _TURNOVER_FLOOR) + fitness = float(sharpe_annual * np.sqrt(abs(ann_return) / denom)) if denom > 0 else 0.0 + + return { + "cumulative_return": cumulative_return, + "sharpe_annual": sharpe_annual, + "turnover_annual": turnover_annual, + "max_drawdown": max_drawdown, + "fitness": fitness, + "hit_rate": float((daily > 0).mean()), + "n_dates": int(len(daily)), + } diff --git a/pipeline/portfolio/simulator.py b/pipeline/portfolio/simulator.py new file mode 100644 index 0000000..38d6308 --- /dev/null +++ b/pipeline/portfolio/simulator.py @@ -0,0 +1,255 @@ +"""Execution simulator: next-open fills under A-share trading constraints. + +Execution model (documented convention): a position book targeted from +information available on date ``t`` is executed at ``open[t+1]``. Trades that +violate a :class:`~pipeline.portfolio.constraints.TradeConstraint` (suspension, +price limit, volume cap, …) are clipped; a fully blocked buy leaves the position +at its previous level. Realized PnL marks the *actually filled* book. + +The simulator is an ABC + a :class:`ReferenceSimulator`; constraints compose by +intersecting their per-name signed delta bounds. +""" + +from __future__ import annotations + +import logging +from abc import ABC, abstractmethod +from dataclasses import dataclass + +import numpy as np +import pandas as pd + +from pipeline.common.schema import FILL_COLUMNS, PNL_COLUMNS +from pipeline.portfolio.constraints import TradeConstraint +from pipeline.portfolio.market_rules import MarketRule, compute_limit_status + +logger = logging.getLogger(__name__) + + +@dataclass +class MarketSlice: + """Per-name market arrays for one execution date (fixed symbol order).""" + + symbol_ids: np.ndarray + date: object + price: np.ndarray # execution/reference price (the open) + preclose: np.ndarray + amount: np.ndarray # daily turnover value (yuan) + tradestatus: np.ndarray # 1 traded / 0 suspended + is_st: np.ndarray + limit_status: np.ndarray # LimitStatus values + close: np.ndarray # close, for marking + + +@dataclass +class TradeContext: + """Inputs handed to constraints and the fill routine for one date.""" + + prev_shares: np.ndarray + target_shares: np.ndarray + slice: MarketSlice + booksize: float + + +@dataclass +class FillResult: + """Outcome of executing one date's target against the constraints.""" + + realized_shares: np.ndarray + traded_shares: np.ndarray + cost: np.ndarray + blocked: np.ndarray + + +class ExecutionSimulator(ABC): + """Abstract execution layer. Subclasses define how a target gets filled.""" + + def __init__(self, constraints: list[TradeConstraint] | None = None, + cost_bps: float = 0.0, slippage_bps: float = 0.0): + self.constraints = constraints or [] + self.cost_bps = cost_bps + self.slippage_bps = slippage_bps + + @abstractmethod + def fill(self, ctx: TradeContext) -> FillResult: + """Execute ``ctx.target_shares`` from ``ctx.prev_shares``.""" + + +class ReferenceSimulator(ExecutionSimulator): + """Reference fill model: clip the desired trade to the composed bounds.""" + + def fill(self, ctx: TradeContext) -> FillResult: + prev = ctx.prev_shares.astype(np.int64) + target = ctx.target_shares.astype(np.int64) + + # Portfolio-level retargeting hooks (e.g. neutrality), if any. + for c in self.constraints: + adjusted = c.adjust_targets(ctx) + if adjusted is not None: + target = np.asarray(adjusted, dtype=np.int64) + + desired = target - prev + n = len(prev) + low = np.full(n, -np.inf) + high = np.full(n, np.inf) + for c in self.constraints: + lo, hi = c.delta_bounds(ctx) + low = np.maximum(low, lo) + high = np.minimum(high, hi) + + # Clip desired delta into the feasible interval; round toward zero so a + # value/volume cap yields a conservative partial fill. + clipped = np.clip(desired.astype(np.float64), low, high) + traded = np.trunc(clipped).astype(np.int64) + blocked = (traded != desired).astype(np.int64) + + realized = prev + traded + open_px = np.where(np.isfinite(ctx.slice.price), ctx.slice.price, 0.0) + trade_value = np.abs(traded.astype(np.float64) * open_px) + cost = trade_value * (self.cost_bps + self.slippage_bps) / 1e4 + return FillResult(realized, traded, cost, blocked) + + def run( + self, + positions_df: pd.DataFrame, + data_df: pd.DataFrame, + rule_engine: MarketRule | None = None, + ) -> tuple[pd.DataFrame, pd.DataFrame]: + """Simulate the whole book date by date with next-open execution. + + For each signal date ``t`` in ``positions_df`` the target is executed at + the *next* available data date's open. Returns ``(fills, pnl)`` with + FILL_COLUMNS / PNL_COLUMNS. + + Args: + positions_df: POSITION_COLUMNS (uses ``target_shares``). + data_df: DATA_COLUMNS (open/close/preclose/amount/tradestatus/isST). + rule_engine: For per-name price-limit bands; default built if None. + + Returns: + ``(fills_df, pnl_df)``. + """ + rule_engine = rule_engine or MarketRule() + portfolio_name = ( + positions_df["portfolio_name"].iloc[0] if len(positions_df) else "" + ) + # Booksize ≈ the per-date gross dollar target (constant by construction). + if "target_value" in positions_df.columns and len(positions_df): + per_date_gross = (positions_df.groupby("date")["target_value"] + .apply(lambda s: s.abs().sum())) + booksize = float(per_date_gross.max()) or 1.0 + else: + booksize = 1.0 + + def wide(df, col): + return df.pivot_table(index="date", columns="symbol_id", + values=col, aggfunc="first").sort_index() + + tgt = wide(positions_df, "target_shares") + opn = wide(data_df, "open") + close = wide(data_df, "close") + preclose = wide(data_df, "preclose") if "preclose" in data_df.columns else close.shift(1) + amount = wide(data_df, "amount") if "amount" in data_df.columns else opn * np.inf + tstat = wide(data_df, "tradestatus") if "tradestatus" in data_df.columns else opn.notna().astype(float) + st = wide(data_df, "isST") if "isST" in data_df.columns else opn * 0.0 + + symbols = sorted(set(tgt.columns) | set(opn.columns)) + tgt = tgt.reindex(columns=symbols) + opn = opn.reindex(columns=symbols) + close = close.reindex(columns=symbols) + preclose = preclose.reindex(columns=symbols) + amount = amount.reindex(columns=symbols) + tstat = tstat.reindex(columns=symbols) + st = st.reindex(columns=symbols) + + sym_arr = np.asarray(symbols, dtype=object) + n = len(symbols) + data_dates = list(close.index) + date_pos = {d: i for i, d in enumerate(data_dates)} + + prev_shares = np.zeros(n, dtype=np.int64) + mark_prev = None # last close at which the book was marked + fill_blocks: list[pd.DataFrame] = [] + pnl_rows: list[dict] = [] + + for t in tgt.index: + # Execute at the next available data date after the signal date t. + i = date_pos.get(t) + if i is None or i + 1 >= len(data_dates): + continue + e = data_dates[i + 1] + + open_e = opn.loc[e].to_numpy(dtype=np.float64) + close_e = close.loc[e].to_numpy(dtype=np.float64) + pre_e = preclose.loc[e].to_numpy(dtype=np.float64) + amt_e = amount.loc[e].to_numpy(dtype=np.float64) + tstat_e = np.nan_to_num(tstat.loc[e].to_numpy(dtype=np.float64), nan=0.0) + st_e = np.nan_to_num(st.loc[e].to_numpy(dtype=np.float64), nan=0.0) + target = np.nan_to_num(tgt.loc[t].to_numpy(dtype=np.float64), nan=0.0).astype(np.int64) + + _, _, _, limit_pct = rule_engine.get_rules_vectorized(sym_arr, e, st_e) + limit_status = compute_limit_status(open_e, pre_e, limit_pct) + + mslice = MarketSlice( + symbol_ids=sym_arr, date=e, price=open_e, preclose=pre_e, + amount=amt_e, tradestatus=tstat_e, is_st=st_e, + limit_status=limit_status, close=close_e, + ) + ctx = TradeContext(prev_shares, target, mslice, booksize) + res = self.fill(ctx) + + # PnL: overnight gap on the OLD book + intraday on the NEW book - cost. + if mark_prev is None: + overnight = 0.0 + else: + gap = np.where(np.isfinite(open_e) & np.isfinite(mark_prev), + open_e - mark_prev, 0.0) + overnight = float(np.nansum(prev_shares * gap)) + intraday_px = np.where(np.isfinite(close_e) & np.isfinite(open_e), + close_e - open_e, 0.0) + intraday = float(np.nansum(res.realized_shares * intraday_px)) + cost_total = float(np.nansum(res.cost)) + pnl = overnight + intraday - cost_total + + mark_e = np.where(np.isfinite(close_e), close_e, open_e) + realized_value = res.realized_shares * np.where(np.isfinite(mark_e), mark_e, 0.0) + traded_value = np.abs(res.traded_shares * np.where(np.isfinite(open_e), open_e, 0.0)) + + nz = res.realized_shares != 0 + fill_blocks.append(pd.DataFrame({ + "symbol_id": symbols, + "date": e, + "portfolio_name": portfolio_name, + "prev_shares": prev_shares, + "target_shares": target, + "traded_shares": res.traded_shares, + "realized_shares": res.realized_shares, + "blocked": res.blocked, + "trade_cost": res.cost, + })[lambda d: (d["traded_shares"] != 0) | (d["realized_shares"] != 0)]) + + pnl_rows.append({ + "date": e, + "portfolio_name": portfolio_name, + "gross_exposure": float(np.abs(realized_value).sum()), + "net_exposure": float(realized_value.sum()), + "pnl": pnl, + "cost": cost_total, + "turnover": float(traded_value.sum() / booksize) if booksize else 0.0, + "n_positions": int(nz.sum()), + }) + + prev_shares = res.realized_shares + mark_prev = mark_e + + fills_df = (pd.concat(fill_blocks, ignore_index=True)[FILL_COLUMNS] + if fill_blocks else pd.DataFrame(columns=FILL_COLUMNS)) + pnl_df = (pd.DataFrame(pnl_rows)[PNL_COLUMNS] + if pnl_rows else pd.DataFrame(columns=PNL_COLUMNS)) + logger.info( + "Simulated '%s': %d exec days, final gross %.0f, total cost %.0f", + portfolio_name, len(pnl_df), + pnl_df["gross_exposure"].iloc[-1] if len(pnl_df) else 0.0, + pnl_df["cost"].sum() if len(pnl_df) else 0.0, + ) + return fills_df, pnl_df diff --git a/tests/test_portfolio.py b/tests/test_portfolio.py new file mode 100644 index 0000000..b50ee99 --- /dev/null +++ b/tests/test_portfolio.py @@ -0,0 +1,394 @@ +"""Tests for the portfolio construction & execution phase (no network).""" + +import datetime as dt + +import numpy as np +import pandas as pd + +from pipeline.common.schema import FILL_COLUMNS, PNL_COLUMNS, POSITION_COLUMNS +from pipeline.portfolio.construct import construct_positions, continuous_targets +from pipeline.portfolio.discretize import repair_exposure, round_to_valid_lot +from pipeline.portfolio.market_rules import ( + Board, + LimitStatus, + MarketRule, + compute_limit_status, + detect_board, +) +from pipeline.portfolio.research import evaluate_portfolio +from pipeline.portfolio.constraints import ( + PriceLimitConstraint, + SuspensionConstraint, + VolumeCapConstraint, +) +from pipeline.portfolio.simulator import ( + MarketSlice, + ReferenceSimulator, + TradeContext, +) + + +# --- fixtures ---------------------------------------------------------------- + +_SYMBOLS = ("sh600000", "sz000001", "sh688981", "sz300750") + + +def _make_data(n_days: int = 40, symbols=_SYMBOLS, start="2024-01-01", + st_symbol=None) -> pd.DataFrame: + """Synthetic long-format DATA_COLUMNS frame, deterministic prices.""" + dates = pd.date_range(start, periods=n_days) + rng = np.random.default_rng(0) + frames = [] + for i, sym in enumerate(symbols): + close = 50.0 + i * 10 + np.cumsum(rng.standard_normal(n_days)) + close = np.abs(close) + 5.0 # keep strictly positive + preclose = np.concatenate([[close[0]], close[:-1]]) + frames.append(pd.DataFrame({ + "symbol_id": sym, + "symbol_name": sym, + "date": dates, + "open": close, + "high": close, + "low": close, + "close": close, + "preclose": preclose, + "volume": 1_000_000.0, + "amount": 1_000_000.0 * close, + "tradestatus": 1, + "isST": 1 if sym == st_symbol else 0, + })) + return pd.concat(frames, ignore_index=True) + + +def _make_weights(data: pd.DataFrame, name="combo") -> pd.DataFrame: + """Demeaned per-date signed weights so the cross-section is dollar-neutral.""" + close = data.pivot_table(index="date", columns="symbol_id", values="close").sort_index() + raw = -close.pct_change(5) + demeaned = raw.sub(raw.mean(axis=1), axis=0) + long = demeaned.reset_index().melt(id_vars="date", var_name="symbol_id", + value_name="weight").dropna() + long["combo_name"] = name + return long[["symbol_id", "date", "combo_name", "weight"]] + + +# --- detect_board ------------------------------------------------------------ + +def test_detect_board(): + assert detect_board("sh600000") == Board.MAIN + assert detect_board("sz000001") == Board.MAIN + assert detect_board("sz002594") == Board.MAIN + assert detect_board("sh688981") == Board.STAR + assert detect_board("sz300750") == Board.CHINEXT + assert detect_board("bj830000") == Board.UNKNOWN + + +# --- MarketRule date transitions --------------------------------------------- + +def test_main_board_increment_transition(): + rules = MarketRule() + before = rules.get_rule("sh600000", dt.date(2023, 8, 9)) + after = rules.get_rule("sh600000", dt.date(2023, 8, 10)) + assert (before.minimum_open_size, before.share_increment) == (100, 100) + assert (after.minimum_open_size, after.share_increment) == (100, 1) + assert before.price_limit_pct == 0.10 + + +def test_star_rule_and_odd_lot(): + rule = MarketRule().get_rule("sh688981", dt.date(2024, 1, 1)) + assert rule.minimum_open_size == 200 + assert rule.share_increment == 1 + assert rule.sell_rule == "odd_lot_full" + assert rule.price_limit_pct == 0.20 + + +def test_st_overrides_price_limit(): + rule = MarketRule().get_rule("sh600000", dt.date(2024, 1, 1), is_st=True) + assert rule.price_limit_pct == 0.05 + + +def test_get_rules_vectorized(): + rules = MarketRule() + syms = np.array(["sh600000", "sh688981", "sz300750"], dtype=object) + min_open, inc, odd, limit = rules.get_rules_vectorized( + syms, dt.date(2024, 1, 1), np.array([0, 0, 0]) + ) + assert list(min_open) == [100, 200, 100] + assert list(inc) == [1, 1, 1] + assert list(odd) == [False, True, False] + assert list(limit) == [0.10, 0.20, 0.20] + + +def test_compute_limit_status(): + price = np.array([110.0, 90.0, 100.0]) + preclose = np.array([100.0, 100.0, 100.0]) + limit_pct = np.array([0.10, 0.10, 0.10]) + status = compute_limit_status(price, preclose, limit_pct) + assert status[0] == LimitStatus.UP_LIMIT.value + assert status[1] == LimitStatus.DOWN_LIMIT.value + assert status[2] == LimitStatus.NORMAL.value + + +# --- continuous targets ------------------------------------------------------ + +def test_continuous_targets_normalization(): + alpha = np.array([2.0, -1.0, -1.0, 0.5]) + price = np.array([10.0, 20.0, 5.0, 8.0]) + w, v_target, q_target = continuous_targets(alpha, price, booksize=1e6) + assert np.isclose(np.abs(w).sum(), 1.0) + assert np.isclose(w.sum(), alpha.sum() / np.abs(alpha).sum()) + assert np.allclose(v_target, 1e6 * w) + assert np.allclose(q_target, v_target / price) + + +def test_continuous_targets_demeaned_is_neutral(): + alpha = np.array([2.0, -1.0, -1.0]) + w, _, _ = continuous_targets(alpha, np.array([10.0, 10.0, 10.0]), 1e6) + assert abs(w.sum()) < 1e-12 + + +def test_continuous_targets_guards_bad_price(): + alpha = np.array([1.0, -1.0]) + w, v, q = continuous_targets(alpha, np.array([np.nan, 10.0]), 1e6) + assert w[0] == 0.0 and q[0] == 0.0 + + +# --- round_to_valid_lot (state-dependent) ------------------------------------ + +def test_round_main_board_pre2023_multiples_of_100(): + target = np.array([250.0, -180.0, 40.0]) + prev = np.zeros(3, dtype=np.int64) + min_open = np.array([100, 100, 100]) + inc = np.array([100, 100, 100]) + out = round_to_valid_lot(target, prev, min_open, inc) + # 250 -> 200 or 300 (nearest is 200? round(150/100)=2 ->300). 250/100 -> k=round(1.5)=2 ->300 + assert out[0] in (200, 300) + assert out[1] in (-200, -100) + assert out[2] == 0 # sub-min, no holding + + +def test_round_post2023_increment_one(): + target = np.array([153.4]) + out = round_to_valid_lot(target, np.zeros(1, np.int64), + np.array([100]), np.array([1])) + assert out[0] == 153 + + +def test_round_star_min_200(): + target = np.array([150.0, 240.6]) + prev = np.zeros(2, dtype=np.int64) + out = round_to_valid_lot(target, prev, np.array([200, 200]), np.array([1, 1]), + np.array([True, True])) + assert out[0] == 0 # below 200, no holding -> cannot open + assert out[1] == 241 # 200 + round(40.6) + + +def test_round_reduction_can_liquidate_below_min(): + # Holding 300, target wants ~40 shares -> nearest valid resting is 0. + target = np.array([40.0]) + prev = np.array([300], dtype=np.int64) + out = round_to_valid_lot(target, prev, np.array([100]), np.array([100])) + assert out[0] == 0 + + +def test_round_star_odd_lot_residual_sells_to_zero(): + # Holding 150 STAR shares (odd lot), target reduces -> must go to 0. + target = np.array([20.0]) + prev = np.array([150], dtype=np.int64) + out = round_to_valid_lot(target, prev, np.array([200]), np.array([1]), + np.array([True])) + assert out[0] == 0 + + +def test_round_no_sign_flip_when_target_same_sign(): + target = np.array([500.0]) + prev = np.array([-300], dtype=np.int64) + out = round_to_valid_lot(target, prev, np.array([100]), np.array([100])) + assert out[0] > 0 # follows target sign, not prev + + +# --- repair_exposure (two-stage) --------------------------------------------- + +def _gross_net(q, price): + v = q.astype(float) * price + return float(np.abs(v).sum()), float(v.sum()) + + +def test_repair_drives_net_and_gross(): + rng = np.random.default_rng(1) + n = 200 + price = rng.uniform(5, 100, n) + alpha = rng.standard_normal(n) + alpha -= alpha.mean() + B = 1e7 + _, _, q_target = continuous_targets(alpha, price, B) + min_open = np.full(n, 100) + inc = np.full(n, 1) + prev = np.zeros(n, dtype=np.int64) + q_round = round_to_valid_lot(q_target, prev, min_open, inc) + pos = repair_exposure(q_round, q_target, price, inc, min_open, prev, + booksize=B, net_tol=0.01, gross_tol=0.01) + gross, net = _gross_net(pos, price) + assert abs(net) <= 0.02 * B + price.max() * 1 # within band + a step + assert abs(gross - B) <= 0.02 * B + price.max() * 1 + + +def test_repair_does_not_worsen_tracking_error_grossly(): + rng = np.random.default_rng(2) + n = 150 + price = rng.uniform(5, 100, n) + alpha = rng.standard_normal(n) + alpha -= alpha.mean() + B = 5e6 + _, v_target, q_target = continuous_targets(alpha, price, B) + inc = np.full(n, 1) + min_open = np.full(n, 100) + prev = np.zeros(n, dtype=np.int64) + q_round = round_to_valid_lot(q_target, prev, min_open, inc) + pos = repair_exposure(q_round, q_target, price, inc, min_open, prev, + booksize=B, net_tol=0.01, gross_tol=0.01) + te_round = np.sum((q_round * price - v_target) ** 2) + te_pos = np.sum((pos * price - v_target) ** 2) + # Repair should keep TE comparable (not blow it up by orders of magnitude). + assert te_pos <= 5.0 * te_round + B + + +def test_repair_scales_to_4000_names(): + rng = np.random.default_rng(3) + n = 4000 + price = rng.uniform(5, 100, n) + alpha = rng.standard_normal(n) + alpha -= alpha.mean() + B = 1e8 + _, _, q_target = continuous_targets(alpha, price, B) + inc = np.full(n, 1) + min_open = np.full(n, 100) + prev = np.zeros(n, dtype=np.int64) + q_round = round_to_valid_lot(q_target, prev, min_open, inc) + pos = repair_exposure(q_round, q_target, price, inc, min_open, prev, booksize=B) + gross, net = _gross_net(pos, price) + assert abs(net) <= 0.03 * B + assert abs(gross - B) <= 0.03 * B + + +# --- construct_positions ----------------------------------------------------- + +def test_construct_positions_schema(): + data = _make_data() + weights = _make_weights(data) + pos = construct_positions(weights, data, booksize=1e6, portfolio_name="run1") + assert list(pos.columns) == POSITION_COLUMNS + assert (pos["portfolio_name"] == "run1").all() + assert pos["position_shares"].dtype == np.int64 + + +def test_construct_positions_threads_state_and_closes_absent(): + data = _make_data() + weights = _make_weights(data) + # Drop the last 3 dates of one symbol so it goes "absent" → must be closed. + sym = "sz300750" + last_dates = sorted(weights["date"].unique())[-3:] + weights = weights[~((weights["symbol_id"] == sym) & + (weights["date"].isin(last_dates)))] + pos = construct_positions(weights, data, booksize=1e6, portfolio_name="run1") + final_date = pos["date"].max() + final = pos[(pos["symbol_id"] == sym) & (pos["date"] == final_date)] + # Either no row, or a zeroed position for the absent name on the final date. + assert final.empty or (final["position_shares"] == 0).all() + + +# --- constraints ------------------------------------------------------------- + +def _slice(n, **over): + base = dict( + symbol_ids=np.array([f"s{i}" for i in range(n)], dtype=object), + date=dt.date(2024, 1, 2), + price=np.full(n, 10.0), + preclose=np.full(n, 10.0), + amount=np.full(n, 1e6), + tradestatus=np.ones(n), + is_st=np.zeros(n), + limit_status=np.zeros(n, dtype=np.int8), + close=np.full(n, 10.0), + ) + base.update(over) + return MarketSlice(**base) + + +def test_suspension_blocks_all_delta(): + n = 2 + sl = _slice(n, tradestatus=np.array([1.0, 0.0])) + ctx = TradeContext(np.zeros(n, np.int64), np.array([100, 100]), sl, 1e6) + low, high = SuspensionConstraint().delta_bounds(ctx) + assert low[1] == 0.0 and high[1] == 0.0 + assert np.isinf(high[0]) + + +def test_price_limit_blocks_directionally(): + n = 2 + sl = _slice(n, limit_status=np.array([LimitStatus.UP_LIMIT.value, + LimitStatus.DOWN_LIMIT.value], dtype=np.int8)) + ctx = TradeContext(np.zeros(n, np.int64), np.array([100, -100]), sl, 1e6) + low, high = PriceLimitConstraint().delta_bounds(ctx) + assert high[0] == 0.0 # up-limit: cannot buy + assert low[1] == 0.0 # down-limit: cannot sell + + +def test_volume_cap_uses_traded_value(): + n = 1 + # amount=1e6, price=10, max_frac=0.1 -> cap value 1e5 -> cap 1e4 shares. + sl = _slice(n, amount=np.array([1e6]), price=np.array([10.0])) + ctx = TradeContext(np.zeros(n, np.int64), np.array([99999]), sl, 1e6) + low, high = VolumeCapConstraint(max_frac=0.1).delta_bounds(ctx) + assert high[0] == 10000.0 + assert low[0] == -10000.0 + + +# --- ReferenceSimulator ------------------------------------------------------ + +def test_simulator_next_open_and_blocked_buy_holds_prev(): + data = _make_data(n_days=15) + weights = _make_weights(data) + pos = construct_positions(weights, data, booksize=1e6, portfolio_name="run1") + sim = ReferenceSimulator(constraints=[SuspensionConstraint()], + cost_bps=5, slippage_bps=5) + fills, pnl = sim.run(pos, data) + assert list(fills.columns) == FILL_COLUMNS + assert list(pnl.columns) == PNL_COLUMNS + # realized = prev + traded must always hold. + assert (fills["realized_shares"] == fills["prev_shares"] + fills["traded_shares"]).all() + + +def test_simulator_blocked_buy_when_suspended(): + n = 1 + sim = ReferenceSimulator(constraints=[SuspensionConstraint()]) + sl = _slice(n, tradestatus=np.array([0.0])) + ctx = TradeContext(np.array([0], np.int64), np.array([500]), sl, 1e6) + res = sim.fill(ctx) + assert res.traded_shares[0] == 0 + assert res.realized_shares[0] == 0 + assert res.blocked[0] == 1 + + +def test_simulator_cost_is_positive_when_trading(): + n = 1 + sim = ReferenceSimulator(constraints=[], cost_bps=10, slippage_bps=5) + sl = _slice(n, price=np.array([20.0])) + ctx = TradeContext(np.array([0], np.int64), np.array([1000]), sl, 1e6) + res = sim.fill(ctx) + assert res.traded_shares[0] == 1000 + # 1000 * 20 * (15/1e4) = 30 + assert np.isclose(res.cost[0], 1000 * 20 * 15 / 1e4) + + +# --- evaluate_portfolio ------------------------------------------------------ + +def test_evaluate_portfolio_keys_no_ic(): + data = _make_data() + weights = _make_weights(data) + pos = construct_positions(weights, data, booksize=1e6, portfolio_name="run1") + metrics = evaluate_portfolio(pos, data) + for key in ("cumulative_return", "sharpe_annual", "turnover_annual", + "max_drawdown", "fitness", "hit_rate", "n_dates"): + assert key in metrics + assert "ic" not in metrics + assert "rank_ic" not in metrics