"""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