diff --git a/apps/backtester/run.py b/apps/backtester/run.py index 23a6686..6920deb 100644 --- a/apps/backtester/run.py +++ b/apps/backtester/run.py @@ -54,6 +54,7 @@ from libs.backtest.earnings_calendar import ( load_pit_earnings_calendar, ) from libs.backtest.form4_calendar import load_pit_form4_calendar +from libs.backtest.ownership_calendar import load_pit_ownership_calendar from libs.backtest.execution import ( simulate_scheduled_open_exit, simulate_entry, @@ -81,6 +82,25 @@ _DIVIDEND_CAPTURE_ENGINE_ID = "idle_dividend_capture" _DIVIDEND_CAPTURE_EVENT_TYPE = "dividend_capture" _FORM4_CAPTURE_ENGINE_ID = "idle_form4_capture" _FORM4_CAPTURE_EVENT_TYPE = "form4_capture" +_OWNERSHIP_CAPTURE_ENGINE_ID = "idle_ownership_13d_capture" +_OWNERSHIP_CAPTURE_EVENT_TYPE = "ownership_13d_capture" +_RISK_OFF_ALPHA_ENGINE_ID = "idle_risk_off_alpha" +_RISK_OFF_ALPHA_EVENT_TYPE = "risk_off_alpha" +_OWNERSHIP_RUNTIME_HOUSEKEEPING_PHRASES = ( + "continued to hold", + "shareholding percentage", + "shareholding percent", + "no amendment to this item", + "change in the number of outstanding", + "number of outstanding shares", + "resulted solely from", + "solely as a result of", + "solely due to", +) +_OWNERSHIP_RUNTIME_STRUCTURAL_PHRASES = ( + "exchange agreement", + "in connection with the reorganization", +) def _get_git_commit_hash() -> str: @@ -153,6 +173,12 @@ class BacktestRunner: if form4_events_path else None ) + ownership_events_path = self.config.ownership_capture.pit_events_path + self._pit_ownership_calendar = ( + load_pit_ownership_calendar(ownership_events_path) + if ownership_events_path + else None + ) self._oracle_pit_earnings_calendar = OraclePointInTimeEarningsCalendar( settings.stock_oracle_url, timeout=float(settings.stock_oracle_timeout), @@ -221,6 +247,7 @@ class BacktestRunner: self._next_trading_day: dict[dt.date, dt.date] = {} self._dividend_capture_trade_counter: int = 0 self._form4_capture_trade_counter: int = 0 + self._ownership_capture_trade_counter: int = 0 def _get_known_upcoming_earnings_by_symbol( self, @@ -271,16 +298,25 @@ class BacktestRunner: return self.initial_equity return self._equity + def _resolve_close_price(self, symbol: str, date: dt.date, fallback: float) -> float: + """Best-effort close price: today's bar → latest prior bar → entry price.""" + bar = self.store.get_bar(symbol, date) + if bar and bar.get("close") is not None and float(bar["close"]) > 0: + return float(bar["close"]) + latest = self.store.get_latest_bar_on_or_before(symbol, date) + if latest is not None: + _, prev_bar = latest + if prev_bar.get("close") is not None and float(prev_bar["close"]) > 0: + return float(prev_bar["close"]) + return fallback + def _compute_portfolio_exposure(self, date: dt.date) -> tuple[float, float]: """Return (gross, net) exposure using current close notional when available.""" gross = 0.0 net = 0.0 for pos in self._open_positions: - bar = self.store.get_bar(pos.plan.candidate.symbol, date) - close = ( - float(bar["close"]) - if bar and bar.get("close") is not None and float(bar["close"]) > 0 - else pos.entry_price + close = self._resolve_close_price( + pos.plan.candidate.symbol, date, pos.entry_price, ) notional = close * pos.shares_open gross += abs(notional) @@ -660,6 +696,9 @@ class BacktestRunner: if self._pending_open_exits.get(date): self._process_pending_open_exits(date) + # --- RISK-OFF ALPHA OPEN EXITS --- + self._process_risk_off_alpha_open_exits(date) + # --- SEASONAL RESET: close stale/underwater positions before peak event season --- if ( self.config.risk.seasonal_reset_enabled @@ -701,7 +740,12 @@ class BacktestRunner: # Kill switch: force close if self._kill_switch_triggered: - trade = simulate_kill_switch_exit(pos, bar, date, self.config.execution) + ks_bar, ks_date = bar, date + if ks_bar is None: + latest = self.store.get_latest_bar_on_or_before(pos.plan.candidate.symbol, date) + if latest is not None: + ks_date, ks_bar = latest + trade = simulate_kill_switch_exit(pos, ks_bar, ks_date, self.config.execution) newly_closed.append(trade) continue @@ -892,6 +936,13 @@ class BacktestRunner: if self._form4_capture_enabled(): self._enter_form4_capture_positions(date) + # --- OWNERSHIP 13D/13G RESIDUAL-CASH SLEEVE --- + if self._ownership_capture_enabled(): + self._enter_ownership_capture_positions(date) + + if self._risk_off_alpha_enabled(): + self._enter_risk_off_alpha_positions(date) + # --- CASH PARKING: invest idle cash --- if self.config.risk.cash_parking_enabled: macro_data_eod = self.store.get_macro_for_date(date) or {} @@ -1059,9 +1110,13 @@ class BacktestRunner: elif target_sym is None: target_sym = park_mode elif target_sym == "sgov" and gate_mode == "volatility": - relay_target = self._evaluate_defensive_relay_target(date, macro_data_eod, park_mode) - if relay_target is not None: - target_sym = relay_target + crisis_target = self._evaluate_crisis_relay_target(macro_data_eod) + if crisis_target is not None: + target_sym = crisis_target + else: + relay_target = self._evaluate_defensive_relay_target(date, macro_data_eod, park_mode) + if relay_target is not None: + target_sym = relay_target # Override for stopped out state if self._parking_stopped_out: target_sym = "sgov" @@ -1295,9 +1350,35 @@ class BacktestRunner: self._commit_parking_target("sgov") self._cash -= new_shares * park_close elif target_sym and investable > 0: + bearish_sym = self.config.risk.cash_parking_bearish_symbol + bearish_alloc_pct = min( + 1.0, + max(0.0, float(self.config.risk.cash_parking_bearish_alloc_pct or 0.0)), + ) + defensive_sym = self._get_parking_defensive_symbol() + defensive_alloc_pct = min( + 1.0, + max(0.0, float(self.config.risk.cash_parking_defensive_alloc_pct or 0.0)), + ) + sgov_amount = 0.0 + symbol_investable = investable + if ( + defensive_sym + and target_sym == defensive_sym + and defensive_alloc_pct < 0.999 + ): + symbol_investable = investable * defensive_alloc_pct + sgov_amount = investable - symbol_investable + elif ( + bearish_sym + and target_sym == bearish_sym + and bearish_alloc_pct < 0.999 + ): + symbol_investable = investable * bearish_alloc_pct + sgov_amount = investable - symbol_investable park_close = macro_data_eod.get(f"{target_sym}_close") - if park_close and park_close > 0 and investable >= park_close: - new_shares = int(investable / park_close) + if park_close and park_close > 0 and symbol_investable >= park_close: + new_shares = int(symbol_investable / park_close) allow_topup = True min_gain_pct = float(self.config.risk.cash_parking_topup_min_gain_pct) if ( @@ -1362,6 +1443,18 @@ class BacktestRunner: macro=macro_data_eod, ) self._cash -= defensive_amount + used_amount = new_shares * park_close if allow_topup else 0.0 + if sgov_amount > 0: + sgov_amount += max(0.0, symbol_investable - used_amount) + elif sgov_amount > 0: + sgov_amount += symbol_investable + if sgov_amount > 0: + self._allocate_parallel_sgov( + date=date, + amount=sgov_amount, + macro=macro_data_eod, + ) + self._cash -= sgov_amount # Track parking entry date and peak price if (self._parking_shares > 0 or self._parking_sgov_value > 0) and self._parking_entry_date is None: @@ -1388,6 +1481,13 @@ class BacktestRunner: else 0.0 ) gross_exposure, net_exposure = self._compute_portfolio_exposure(date) + ia_exposure_val = 0.0 + for _pos in self._open_positions: + if self._is_post_allocation_idle_engine_id(_pos.plan.engine_id): + _close = self._resolve_close_price( + _pos.plan.candidate.symbol, date, _pos.entry_price, + ) + ia_exposure_val += _close * _pos.shares_open self._equity_curve.append( DailyPortfolioState( date=date, @@ -1403,6 +1503,10 @@ class BacktestRunner: daily_new_risk_used=self._daily_new_risk_used, peak_equity=self._peak_equity, current_drawdown_pct=final_drawdown, + raw_cash=self._cash, + parking_value=parking_value, + idle_alpha_exposure=ia_exposure_val, + primary_exposure=max(0.0, gross_exposure - ia_exposure_val), ) ) @@ -2574,6 +2678,282 @@ class BacktestRunner: and self._pit_form4_calendar is not None ) + def _ownership_capture_enabled(self) -> bool: + cfg = self.config.ownership_capture + return bool( + cfg.enabled + and cfg.max_positions > 0 + and cfg.max_new_per_day > 0 + and cfg.hold_days > 0 + and (cfg.max_idle_deploy_pct > 0 or cfg.reserve_pct > 0) + and self._pit_ownership_calendar is not None + ) + + def _risk_off_alpha_enabled(self) -> bool: + cfg = self.config.risk_off_alpha + return bool( + cfg.enabled + and len(cfg.symbols) > 0 + and (float(cfg.reserve_pct) > 0 or float(cfg.max_idle_deploy_pct) > 0) + ) + + def _preview_effective_parking_target(self, date: dt.date) -> str | None: + macro = self.store.get_macro_for_date(date) or {} + if not macro: + return None + trend_state_before = self._parking_trend_sgov + gate_state_before = self._parking_gate_in_sgov + raw_target = self._compute_parking_target(date) + self._parking_trend_sgov = trend_state_before + self._parking_gate_in_sgov = gate_state_before + target = raw_target + if target == "sgov" and self.config.risk.cash_parking_gate_mode == "volatility": + crisis_target = self._evaluate_crisis_relay_target(macro) + if crisis_target is not None: + target = crisis_target + else: + relay_target = self._evaluate_defensive_relay_target( + date, + macro, + self.config.risk.cash_parking_symbol, + ) + if relay_target is not None: + target = relay_target + bearish_sym = self.config.risk.cash_parking_bearish_symbol + if bearish_sym and target == "sgov": + risk_score = self._compute_parking_risk_score(macro) + if risk_score >= self.config.risk.cash_parking_bearish_threshold: + target = bearish_sym + return target + + def _risk_off_alpha_sgov_streak(self, signal_date: dt.date) -> int: + streak = 0 + cursor = signal_date + while cursor is not None: + if self._preview_effective_parking_target(cursor) != "sgov": + break + streak += 1 + cursor = self._previous_simulation_date(cursor) + return streak + + def _select_risk_off_alpha_symbol( + self, + signal_date: dt.date, + *, + current_symbol: str | None = None, + ) -> str | None: + if not self._risk_off_alpha_enabled(): + return None + if self._preview_effective_parking_target(signal_date) != "sgov": + return None + + cfg = self.config.risk_off_alpha + if self._risk_off_alpha_sgov_streak(signal_date) < int(cfg.min_consecutive_sgov_days): + return None + + macro = self.store.get_macro_for_date(signal_date) or {} + if not macro: + return None + if float(cfg.min_parking_risk_score or 0.0) > 0: + risk_score = float(self._compute_parking_risk_score(macro)) + if risk_score < float(cfg.min_parking_risk_score): + return None + + lookback_days = max(1, int(cfg.momentum_lookback_days or 20)) + min_mom = float(cfg.min_symbol_momentum or 0.0) + candidates: list[tuple[str, float]] = [] + for symbol in cfg.symbols: + normalized = str(symbol).lower().strip() + if not normalized: + continue + prefix = self._get_parking_signal_prefix(normalized) + momentum = macro.get(f"{prefix}_mom_{lookback_days}") + if momentum is None or float(momentum) < min_mom: + continue + candidates.append((normalized, float(momentum))) + if not candidates: + return None + candidates.sort(key=lambda item: item[1], reverse=True) + best_symbol, best_score = candidates[0] + + if current_symbol: + current_normalized = str(current_symbol).lower() + current_score = next((score for symbol, score in candidates if symbol == current_normalized), None) + gap = float(cfg.rotation_momentum_gap or 0.0) + if current_score is not None and best_symbol != current_normalized and best_score <= current_score + gap: + return current_normalized + return best_symbol + + def _build_risk_off_alpha_candidate( + self, + *, + symbol: str, + date: dt.date, + signal_date: dt.date, + signal_momentum: float, + sgov_streak: int, + avg_dollar_volume: float, + ) -> Candidate: + event_timestamp = dt.datetime.combine(signal_date, dt.time(0, 0), tzinfo=dt.timezone.utc) + return Candidate( + event_id=f"{_RISK_OFF_ALPHA_EVENT_TYPE}:{symbol}:{signal_date.isoformat()}", + symbol=symbol.upper(), + source_symbol=symbol.upper(), + score=float(signal_momentum) * 1_000_000.0 + float(sgov_streak), + sector="MACRO", + event_type=_RISK_OFF_ALPHA_EVENT_TYPE, + event_timestamp=event_timestamp, + event_date=signal_date, + filing_time_bucket="unknown", + timing_class="unknown", + reaction_date=date, + execution_date=date, + entry_price_est=float((self.store.get_bar(symbol.upper(), date) or {}).get("open") or 0.0), + avg_dollar_volume=avg_dollar_volume, + score_bucket="idle_risk_off_alpha", + engine_id=_RISK_OFF_ALPHA_ENGINE_ID, + entry_timing_policy="next_open", + engine_max_holding_days=int(self.config.risk_off_alpha.max_holding_days or 0) or None, + features={ + "trade_sleeve": "risk_off_alpha", + "risk_off_alpha_signal_date": signal_date.isoformat(), + "risk_off_alpha_symbol": symbol.lower(), + "risk_off_alpha_signal_momentum": signal_momentum, + "risk_off_alpha_sgov_streak": sgov_streak, + }, + ) + + def _process_risk_off_alpha_open_exits(self, date: dt.date) -> None: + if not self._open_positions: + return + signal_date = self._previous_simulation_date(date) + if signal_date is None: + return + + remaining_positions: list[OpenPosition] = [] + max_hold_days = max(0, int(self.config.risk_off_alpha.max_holding_days or 0)) + for position in self._open_positions: + if position.plan.engine_id != _RISK_OFF_ALPHA_ENGINE_ID: + remaining_positions.append(position) + continue + + symbol = str(position.plan.candidate.symbol).upper() + desired_symbol = self._select_risk_off_alpha_symbol(signal_date, current_symbol=symbol) + exit_reason: str | None = None + if desired_symbol is None: + exit_reason = "RISK_OFF_ALPHA_OFF" + elif str(desired_symbol).upper() != symbol: + exit_reason = "RISK_OFF_ALPHA_ROTATE" + elif max_hold_days > 0 and position.days_held >= max_hold_days: + exit_reason = "RISK_OFF_ALPHA_MAX_HOLD" + + if exit_reason is None: + remaining_positions.append(position) + continue + + bar = self.store.get_bar(symbol, date) + if bar is None or bar.get("open") is None: + remaining_positions.append(position) + continue + trade = simulate_scheduled_open_exit( + position=position, + bar=bar, + config=self._build_effective_execution_config(position.plan.candidate), + current_date=date, + reason=exit_reason, + fraction=1.0, + ) + if trade is None: + remaining_positions.append(position) + continue + self._closed_trades.append(trade) + self._candidate_map[trade.trade_id] = position.plan.candidate + self._realized_pnl += trade.net_pnl + self._cash += trade.net_pnl + (trade.entry_price * trade.shares) + if trade.net_pnl < 0: + self._consecutive_losses += 1 + else: + self._consecutive_losses = 0 + self._open_positions = remaining_positions + + def _enter_risk_off_alpha_positions(self, date: dt.date) -> None: + signal_date = self._previous_simulation_date(date) + if signal_date is None: + return + desired_symbol = self._select_risk_off_alpha_symbol(signal_date) + if desired_symbol is None: + return + + existing_positions = [ + position for position in self._open_positions + if position.plan.engine_id == _RISK_OFF_ALPHA_ENGINE_ID + ] + if any(str(position.plan.candidate.symbol).upper() == str(desired_symbol).upper() for position in existing_positions): + return + if existing_positions: + return + + cfg = self.config.risk_off_alpha + market_value = self._compute_positions_market_value(date) + equity_est = self._cash + market_value + self._get_parking_value(date) + if equity_est <= 0: + return + if float(cfg.reserve_pct) > 0: + reserve_budget = equity_est * float(cfg.reserve_pct) + if reserve_budget <= 0: + return + if self._cash + 1e-9 < reserve_budget and self._get_parking_value(date) > 0: + self._liquidate_parking_for_cash(date, reserve_budget - self._cash) + total_budget = min(reserve_budget, self._cash) + else: + cash_ratio = self._cash / equity_est + if cash_ratio < float(cfg.min_cash_ratio_for_overlay): + return + total_budget = self._cash * min(1.0, max(0.0, float(cfg.max_idle_deploy_pct))) + if total_budget <= 0: + return + + symbol = str(desired_symbol).upper() + entry_bar = self.store.get_bar(symbol, date) + if entry_bar is None or entry_bar.get("open") is None: + return + avg_dollar_volume = float(self.store.get_market_features(symbol, date).get("avg_dollar_volume_20d") or 0.0) + signal_macro = self.store.get_macro_for_date(signal_date) or {} + signal_momentum = float(signal_macro.get(f"{symbol.lower()}_mom_{max(1, int(cfg.momentum_lookback_days or 20))}") or 0.0) + candidate = self._build_risk_off_alpha_candidate( + symbol=symbol, + date=date, + signal_date=signal_date, + signal_momentum=signal_momentum, + sgov_streak=self._risk_off_alpha_sgov_streak(signal_date), + avg_dollar_volume=avg_dollar_volume, + ) + effective_exec = self._build_effective_execution_config(candidate) + estimated_fill = float(entry_bar["open"]) * (1.0 + effective_exec.slippage_bps_base / 10_000.0) + shares = int(total_budget / estimated_fill) if estimated_fill > 0 else 0 + if shares <= 0: + return + plan = PlannedOrder( + candidate=candidate, + shares=shares, + entry_price_limit=float(entry_bar["open"]), + stop_price=0.01, + target_price=float(entry_bar["open"]) * 100.0, + risk_dollars=0.0, + event_date=signal_date, + timing_class="unknown", + engine_id=_RISK_OFF_ALPHA_ENGINE_ID, + entry_timing_policy="next_open", + ) + position = simulate_entry(plan, entry_bar, effective_exec) + if position is None: + return + trade_cost = position.entry_price * position.shares_open + if trade_cost <= 0 or trade_cost > self._cash + 1e-9: + return + self._cash -= trade_cost + self._open_positions.append(position) + def _process_dividend_capture_open_exits(self, date: dt.date) -> None: if not self._open_positions: return @@ -2916,6 +3296,12 @@ class BacktestRunner: score_bucket="idle_form4", engine_id=_FORM4_CAPTURE_ENGINE_ID, entry_timing_policy="next_open", + engine_early_failure_close_below_entry_and_reaction_close=( + not bool(self.config.form4_capture.disable_day1_early_failure) + ), + engine_early_failure_no_progress_days=self.config.form4_capture.no_progress_days_override, + engine_early_failure_no_progress_r=self.config.form4_capture.no_progress_r_override, + engine_early_failure_no_progress_fraction=self.config.form4_capture.no_progress_fraction_override, engine_max_holding_days=int(self.config.form4_capture.hold_days), features={ "trade_sleeve": "idle_alpha", @@ -3000,6 +3386,355 @@ class BacktestRunner: self._cash -= trade_cost self._open_positions.append(position) + def _select_ownership_capture_candidates(self, date: dt.date) -> list[dict[str, Any]]: + if not self._ownership_capture_enabled(): + return [] + + date_index = self._simulation_date_index.get(date) + if date_index is None or date_index <= 0: + return [] + + prev_trading_date = self._simulation_dates[date_index - 1] + start_filing_date = prev_trading_date + end_filing_date = date - dt.timedelta(days=1) + if end_filing_date < start_filing_date: + return [] + + cfg = self.config.ownership_capture + filing_events = self._pit_ownership_calendar.get_events_between( + start_filing_date=start_filing_date, + end_filing_date=end_filing_date, + symbols=self.store._bars.keys(), + ) + if not filing_events: + return [] + + allowed_form_groups = {str(group).strip().upper() for group in cfg.form_groups if str(group).strip()} + open_symbols = { + str(position.plan.candidate.symbol).upper() + for position in self._open_positions + } + if self._parking_current_symbol: + open_symbols.add(str(self._parking_current_symbol).upper()) + + best_by_symbol: dict[str, dict[str, Any]] = {} + loss_cooldown_days = max(0, int(cfg.symbol_cooldown_days_after_loss or 0)) + symbol_max_entries = ( + int(cfg.symbol_max_entries_in_lookback) + if cfg.symbol_max_entries_in_lookback is not None + else 0 + ) + symbol_entry_lookback_days = max(1, int(cfg.symbol_entry_lookback_days or 365)) + for event in filing_events: + if self._first_trading_day_after(event.filing_date) != date: + continue + form_group = "13D" if "13D" in str(event.form_type).upper() else "13G" + if allowed_form_groups and form_group not in allowed_form_groups: + continue + if bool(cfg.require_amendment) and not bool(event.is_amendment): + continue + if bool(cfg.require_initial) and not bool(event.is_initial_for_owner): + continue + if bool(cfg.require_activist) and not bool(event.activist_flag): + continue + if bool(cfg.require_13g_to_13d_transition) and not bool(event.is_13g_to_13d_transition): + continue + if float(event.percent_owned or 0.0) < float(cfg.min_percent_owned): + continue + if float(cfg.min_percent_delta_points) > 0 and float(event.percent_delta_points or 0.0) < float(cfg.min_percent_delta_points): + continue + purpose_text = str(event.purpose_text or "") + runtime_housekeeping = False + runtime_structural = False + if purpose_text: + lowered_purpose = purpose_text.lower() + runtime_housekeeping = any( + phrase in lowered_purpose for phrase in _OWNERSHIP_RUNTIME_HOUSEKEEPING_PHRASES + ) + runtime_structural = any( + phrase in lowered_purpose for phrase in _OWNERSHIP_RUNTIME_STRUCTURAL_PHRASES + ) + if bool(cfg.exclude_housekeeping_purpose) and ( + bool(event.purpose_housekeeping_flag) or runtime_housekeeping + ): + continue + if bool(cfg.exclude_structural_exchange_purpose) and runtime_structural: + continue + if ( + cfg.min_strength_score is not None + and int(event.ownership_strength_score or 0) < int(cfg.min_strength_score) + ): + continue + + symbol = str(event.symbol).upper() + if symbol in open_symbols: + continue + if loss_cooldown_days > 0: + recent_loss_found = False + for trade in self._closed_trades: + candidate = self._candidate_map.get(trade.trade_id) + if candidate is None or candidate.engine_id != _OWNERSHIP_CAPTURE_ENGINE_ID: + continue + if str(candidate.symbol).upper() != symbol: + continue + if float(trade.net_pnl) >= 0: + continue + if (date - trade.exit_date).days <= loss_cooldown_days: + recent_loss_found = True + break + if recent_loss_found: + continue + if symbol_max_entries > 0: + recent_entry_count = 0 + for trade in self._closed_trades: + candidate = self._candidate_map.get(trade.trade_id) + if candidate is None or candidate.engine_id != _OWNERSHIP_CAPTURE_ENGINE_ID: + continue + if str(candidate.symbol).upper() != symbol: + continue + if (date - trade.entry_date).days > symbol_entry_lookback_days: + continue + recent_entry_count += 1 + if recent_entry_count >= symbol_max_entries: + break + if recent_entry_count >= symbol_max_entries: + continue + bar = self.store.get_bar(symbol, date) + if bar is None or bar.get("open") is None: + continue + features = self.store.get_market_features(symbol, date) + avg_dollar_volume = float(features.get("avg_dollar_volume_20d") or 0.0) + if avg_dollar_volume < float(cfg.min_avg_dollar_volume): + continue + sort_key = ( + int(event.ownership_strength_score or 0), + 1 if bool(event.activist_flag) else 0, + float(event.percent_delta_points or 0.0), + float(event.percent_owned or 0.0), + ) + row = { + "symbol": symbol, + "filing_date": event.filing_date, + "form_type": str(event.form_type).upper(), + "owner_name": event.owner_name, + "owner_key": event.owner_key, + "percent_owned": float(event.percent_owned or 0.0), + "aggregate_shares": float(event.aggregate_shares or 0.0), + "purpose_text": event.purpose_text, + "purpose_housekeeping_flag": bool(event.purpose_housekeeping_flag or runtime_housekeeping), + "ownership_structural_exchange_flag": bool(runtime_structural), + "activist_flag": bool(event.activist_flag), + "is_amendment": bool(event.is_amendment), + "prior_percent_owned": ( + float(event.prior_percent_owned) + if event.prior_percent_owned is not None + else None + ), + "percent_delta_points": ( + float(event.percent_delta_points) + if event.percent_delta_points is not None + else None + ), + "prior_form_group": event.prior_form_group, + "is_initial_for_owner": bool(event.is_initial_for_owner), + "is_13g_to_13d_transition": bool(event.is_13g_to_13d_transition), + "ownership_strength_score": int(event.ownership_strength_score or 0), + "avg_dollar_volume": avg_dollar_volume, + "_sort": sort_key, + } + existing = best_by_symbol.get(symbol) + if existing is None or tuple(row["_sort"]) > tuple(existing["_sort"]): + best_by_symbol[symbol] = row + + rows = list(best_by_symbol.values()) + rows.sort(key=lambda item: item["_sort"], reverse=True) + for row in rows: + row.pop("_sort", None) + return rows + + def _build_ownership_capture_candidate( + self, + *, + symbol: str, + date: dt.date, + filing_date: dt.date, + form_type: str, + percent_owned: float, + aggregate_shares: float, + activist_flag: bool, + is_amendment: bool, + prior_percent_owned: float | None, + percent_delta_points: float | None, + prior_form_group: str | None, + is_initial_for_owner: bool, + is_13g_to_13d_transition: bool, + avg_dollar_volume: float, + owner_name: str | None, + owner_key: str | None, + purpose_text: str | None, + purpose_housekeeping_flag: bool, + ownership_strength_score: int, + ) -> Candidate: + event_timestamp = dt.datetime.combine(filing_date, dt.time(0, 0), tzinfo=dt.timezone.utc) + score = ( + (1_000_000_000.0 if activist_flag else 0.0) + + float(percent_delta_points or 0.0) * 100_000_000.0 + + float(percent_owned or 0.0) * 1_000_000.0 + ) + return Candidate( + event_id=f"{_OWNERSHIP_CAPTURE_EVENT_TYPE}:{symbol}:{filing_date.isoformat()}:{form_type}", + symbol=symbol, + source_symbol=symbol, + score=score, + sector="OWNERSHIP", + event_type=_OWNERSHIP_CAPTURE_EVENT_TYPE, + event_timestamp=event_timestamp, + event_date=filing_date, + filing_time_bucket="unknown", + timing_class="unknown", + reaction_date=date, + execution_date=date, + entry_price_est=float((self.store.get_bar(symbol, date) or {}).get("open") or 0.0), + avg_dollar_volume=avg_dollar_volume, + score_bucket="idle_ownership", + engine_id=_OWNERSHIP_CAPTURE_ENGINE_ID, + entry_timing_policy="next_open", + engine_early_failure_close_below_entry_and_reaction_close=( + not bool(self.config.ownership_capture.disable_day1_early_failure) + ), + engine_early_failure_no_progress_days=self.config.ownership_capture.no_progress_days_override, + engine_early_failure_no_progress_r=self.config.ownership_capture.no_progress_r_override, + engine_early_failure_no_progress_fraction=self.config.ownership_capture.no_progress_fraction_override, + engine_max_holding_days=int(self.config.ownership_capture.hold_days), + features={ + "trade_sleeve": "ownership", + "ownership_filing_date": filing_date.isoformat(), + "ownership_form_type": form_type, + "ownership_percent_owned": percent_owned, + "ownership_aggregate_shares": aggregate_shares, + "ownership_purpose_text": purpose_text, + "ownership_purpose_housekeeping_flag": purpose_housekeeping_flag, + "ownership_activist_flag": activist_flag, + "ownership_is_amendment": is_amendment, + "ownership_prior_percent_owned": prior_percent_owned, + "ownership_percent_delta_points": percent_delta_points, + "ownership_prior_form_group": prior_form_group, + "ownership_is_initial_for_owner": is_initial_for_owner, + "ownership_is_13g_to_13d_transition": is_13g_to_13d_transition, + "ownership_strength_score": ownership_strength_score, + "ownership_owner_name": owner_name, + "ownership_owner_key": owner_key, + }, + ) + + def _enter_ownership_capture_positions(self, date: dt.date) -> None: + selected = self._select_ownership_capture_candidates(date) + if not selected: + return + + cfg = self.config.ownership_capture + existing_positions = sum( + 1 for position in self._open_positions + if position.plan.engine_id == _OWNERSHIP_CAPTURE_ENGINE_ID + ) + available_slots = max(0, int(cfg.max_positions) - existing_positions) + if available_slots <= 0: + return + + selected = selected[: min(int(cfg.max_new_per_day), available_slots)] + if not selected: + return + + market_value = self._compute_positions_market_value(date) + equity_est = self._cash + market_value + self._get_parking_value(date) + if equity_est <= 0: + return + if float(cfg.reserve_pct) > 0: + reserve_budget = equity_est * float(cfg.reserve_pct) + if reserve_budget <= 0: + return + if self._cash + 1e-9 < reserve_budget and self._get_parking_value(date) > 0: + self._liquidate_parking_for_cash(date, reserve_budget - self._cash) + total_budget = min(reserve_budget, self._cash) + extra_idle_deploy_pct = min( + 1.0, + max(0.0, float(cfg.extra_idle_deploy_pct_above_reserve or 0.0)), + ) + if extra_idle_deploy_pct > 0 and equity_est > 0: + cash_ratio = self._cash / equity_est + if cash_ratio >= float(cfg.min_cash_ratio_for_overlay): + remaining_cash = max(0.0, self._cash - total_budget) + total_budget += remaining_cash * extra_idle_deploy_pct + else: + cash_ratio = self._cash / equity_est + if cash_ratio < float(cfg.min_cash_ratio_for_overlay): + return + total_budget = self._cash * min(1.0, max(0.0, float(cfg.max_idle_deploy_pct))) + if total_budget <= 0: + return + + selected_candidates: list[tuple[dict[str, Any], Candidate]] = [] + for payload in selected: + symbol = str(payload["symbol"]) + candidate = self._build_ownership_capture_candidate( + symbol=symbol, + date=date, + filing_date=payload["filing_date"], + form_type=str(payload["form_type"]), + percent_owned=float(payload["percent_owned"]), + aggregate_shares=float(payload["aggregate_shares"]), + activist_flag=bool(payload["activist_flag"]), + is_amendment=bool(payload["is_amendment"]), + prior_percent_owned=payload["prior_percent_owned"], + percent_delta_points=payload["percent_delta_points"], + prior_form_group=payload["prior_form_group"], + is_initial_for_owner=bool(payload["is_initial_for_owner"]), + is_13g_to_13d_transition=bool(payload["is_13g_to_13d_transition"]), + avg_dollar_volume=float(payload["avg_dollar_volume"]), + owner_name=payload.get("owner_name"), + owner_key=payload.get("owner_key"), + purpose_text=payload.get("purpose_text"), + purpose_housekeeping_flag=bool(payload.get("purpose_housekeeping_flag")), + ownership_strength_score=int(payload.get("ownership_strength_score") or 0), + ) + selected_candidates.append((payload, candidate)) + + if not selected_candidates: + return + + per_position_budgets = [total_budget / len(selected_candidates)] * len(selected_candidates) + + for (payload, candidate), per_position_budget in zip(selected_candidates, per_position_budgets, strict=False): + symbol = str(payload["symbol"]) + entry_bar = self.store.get_bar(symbol, date) + if entry_bar is None or entry_bar.get("open") is None: + continue + effective_exec = self._build_effective_execution_config(candidate) + estimated_fill = float(entry_bar["open"]) * (1.0 + effective_exec.slippage_bps_base / 10_000.0) + shares = int(per_position_budget / estimated_fill) if estimated_fill > 0 else 0 + if shares <= 0: + continue + plan = PlannedOrder( + candidate=candidate, + shares=shares, + entry_price_limit=float(entry_bar["open"]), + stop_price=0.01, + target_price=float(entry_bar["open"]) * 100.0, + risk_dollars=0.0, + event_date=payload["filing_date"], + timing_class="unknown", + engine_id=_OWNERSHIP_CAPTURE_ENGINE_ID, + entry_timing_policy="next_open", + ) + position = simulate_entry(plan, entry_bar, effective_exec) + if position is None: + continue + trade_cost = position.entry_price * position.shares_open + if trade_cost <= 0 or trade_cost > self._cash + 1e-9: + continue + self._cash -= trade_cost + self._open_positions.append(position) + def _schedule_add_on_candidates(self, date: dt.date) -> None: next_date = self._next_trading_day.get(date) if next_date is None: @@ -4085,16 +4820,14 @@ class BacktestRunner: """ total = 0.0 for pos in self._open_positions: - bar = self.store.get_bar(pos.plan.candidate.symbol, date) + close = self._resolve_close_price( + pos.plan.candidate.symbol, date, pos.entry_price, + ) is_short = pos.plan.candidate.trade_direction == "short" - if bar and bar.get("close"): - close = float(bar["close"]) - if is_short: - total += (2.0 * pos.entry_price - close) * pos.shares_open - else: - total += close * pos.shares_open + if is_short: + total += (2.0 * pos.entry_price - close) * pos.shares_open else: - total += pos.entry_price * pos.shares_open + total += close * pos.shares_open return total def _compute_unrealized_pnl(self, date: dt.date) -> float: @@ -4106,17 +4839,85 @@ class BacktestRunner: def _get_parking_signal_prefix(self, symbol: str | None) -> str: """Map parking symbol to the macro prefix that has full signal coverage.""" normalized = (symbol or "").lower() - if normalized in ("spy", "spym", "qqq"): + # QQQM is a lower-fee parking vehicle, but the macro feature store only + # computes the full signal stack (mom/vol/entropy/autocorr/...) for QQQ. + # If we use `qqqm` as the signal prefix, stress exits can still fire from + # generic QQQ features, but SGOV -> QQQM re-entry confirmation never + # resolves because `qqqm_mom_*` etc. are missing. That leaves parking + # effectively stuck in SGOV after the first risk-off transition. + if normalized == "qqqm": + return "qqq" + if normalized in ( + "spy", + "spym", + "qqq", + "qual", + "gld", + "jepq", + "bufb", + "merix", + "shy", + "usfr", + "bil", + "vgsh", + "iei", + "ief", + "tip", + "dbc", + "sh", + "psq", + ): return normalized return "qqq" def _get_parking_defensive_symbol(self) -> str: """Secondary defensive ETF used for multi-step parking ladders.""" symbol = (self.config.risk.cash_parking_defensive_symbol or "spy").lower() - if symbol in ("spy", "spym", "qual"): + if symbol in ( + "spy", + "spym", + "qual", + "gld", + "dbc", + "jepq", + "bufb", + "merix", + "shy", + "usfr", + "bil", + "vgsh", + "iei", + "ief", + "tip", + "dbc", + ): return symbol return "spy" + def _get_parking_defensive_alt_symbol(self) -> str | None: + symbol = (self.config.risk.cash_parking_defensive_alt_symbol or "").lower() + if not symbol: + return None + if symbol in ( + "spy", + "spym", + "qual", + "gld", + "dbc", + "jepq", + "bufb", + "merix", + "shy", + "usfr", + "bil", + "vgsh", + "iei", + "ief", + "tip", + ): + return symbol + return None + def _get_parking_defensive_prefix(self) -> str: defensive_symbol = self._get_parking_defensive_symbol() if defensive_symbol == "qual": @@ -4126,6 +4927,36 @@ class BacktestRunner: def _get_parking_defensive_corr_key(self) -> str: return f"{self._get_parking_defensive_prefix()}_qqq_corr_20" + def _get_parking_crisis_symbol(self) -> str | None: + symbol = (self.config.risk.cash_parking_crisis_symbol or "").lower() + if symbol in ("gld", "shy", "usfr", "bil", "vgsh", "iei", "ief", "tip", "dbc"): + return symbol + return None + + def _evaluate_crisis_relay_target(self, macro: dict) -> str | None: + """Use a bond-like safe haven only during deep stress and only when it is already trending up.""" + crisis_symbol = self._get_parking_crisis_symbol() + if crisis_symbol is None: + return None + + risk_score = float(self._compute_parking_risk_score(macro)) + if risk_score < float(self.config.risk.cash_parking_crisis_threshold): + return None + + prefix = self._get_parking_signal_prefix(crisis_symbol) + mom_days = max(1, int(self.config.risk.cash_parking_crisis_momentum_days or 20)) + momentum = macro.get(f"{prefix}_mom_{mom_days}") + if momentum is None or momentum <= float(self.config.risk.cash_parking_crisis_momentum_min): + return None + + vol_max = float(self.config.risk.cash_parking_crisis_vol_max or 0.0) + if vol_max > 0: + vol = macro.get(f"{prefix}_vol_20") + if vol is None or vol > vol_max: + return None + + return crisis_symbol + def _evaluate_defensive_relay_target( self, date: dt.date, @@ -4137,14 +4968,8 @@ class BacktestRunner: return None defensive_symbol = self._get_parking_defensive_symbol() - if defensive_symbol in ("", "sgov", park_mode): - return None - - defensive_prefix = self._get_parking_defensive_prefix() signal_prefix = self._get_parking_signal_prefix(park_mode) period = self.config.risk.cash_parking_trend_sma_period - vol_lb = self.config.risk.cash_parking_gate_vol_lookback - ent_lb = self.config.risk.cash_parking_entropy_lookback risk_score = float(self._compute_parking_risk_score(macro)) relay_risk_cap = min( @@ -4175,37 +5000,58 @@ class BacktestRunner: if trigger_mode == "turn_or_recovery" and not (turn_ok or recovery_ok): return None - defensive_vol_threshold = ( - self.config.risk.cash_parking_gate_vol_spy_threshold - or self.config.risk.cash_parking_gate_vol_threshold - ) - defensive_vol = macro.get(f"{defensive_prefix}_vol_{vol_lb}") - if defensive_vol is not None and defensive_vol >= defensive_vol_threshold: - return None - - defensive_mom = macro.get(f"{defensive_prefix}_mom_{period}") - if defensive_mom is not None and defensive_mom <= -0.01: - return None - - ent_thr = self.config.risk.cash_parking_entropy_threshold - if ent_thr > 0: - defensive_entropy = macro.get(f"{defensive_prefix}_entropy_{ent_lb}") - if defensive_entropy is not None and defensive_entropy > ent_thr + 0.15: - return None + def _candidate_ok(symbol: str) -> tuple[bool, float]: + if symbol in ("", "sgov", park_mode): + return False, float("-inf") + prefix = self._get_parking_signal_prefix(symbol) + vol_lb = self.config.risk.cash_parking_gate_vol_lookback + ent_lb = self.config.risk.cash_parking_entropy_lookback + defensive_vol_threshold = float( + self.config.risk.cash_parking_defensive_vol_max + or self.config.risk.cash_parking_gate_vol_spy_threshold + or self.config.risk.cash_parking_gate_vol_threshold + ) + defensive_vol = macro.get(f"{prefix}_vol_{vol_lb}") + if defensive_vol is not None and defensive_vol >= defensive_vol_threshold: + return False, float("-inf") - defensive_downside = macro.get(f"{defensive_prefix}_downside_vol_20") - if defensive_downside is not None and defensive_downside > 0.18: - return None + defensive_mom = macro.get(f"{prefix}_mom_{period}") + if defensive_mom is not None and defensive_mom <= float(self.config.risk.cash_parking_defensive_momentum_min): + return False, float("-inf") - defensive_ulcer = macro.get(f"{defensive_prefix}_ulcer_20") - if defensive_ulcer is not None and defensive_ulcer > 0.06: - return None - - defensive_autocorr = macro.get(f"{defensive_prefix}_autocorr_20") - if defensive_autocorr is not None and defensive_autocorr < -0.10: + ent_thr = self.config.risk.cash_parking_entropy_threshold + if ent_thr > 0: + defensive_entropy = macro.get(f"{prefix}_entropy_{ent_lb}") + if defensive_entropy is not None and defensive_entropy > ent_thr + 0.15: + return False, float("-inf") + + defensive_downside = macro.get(f"{prefix}_downside_vol_20") + if defensive_downside is not None and defensive_downside > 0.18: + return False, float("-inf") + + defensive_ulcer = macro.get(f"{prefix}_ulcer_20") + if defensive_ulcer is not None and defensive_ulcer > 0.06: + return False, float("-inf") + + defensive_autocorr = macro.get(f"{prefix}_autocorr_20") + if defensive_autocorr is not None and defensive_autocorr < -0.10: + return False, float("-inf") + + return True, float(defensive_mom or 0.0) + + candidates: list[tuple[str, float]] = [] + ok, score = _candidate_ok(defensive_symbol) + if ok: + candidates.append((defensive_symbol, score)) + alt_symbol = self._get_parking_defensive_alt_symbol() + if alt_symbol and alt_symbol != defensive_symbol: + ok, score = _candidate_ok(alt_symbol) + if ok: + candidates.append((alt_symbol, score)) + if not candidates: return None - - return defensive_symbol + candidates.sort(key=lambda item: item[1], reverse=True) + return candidates[0][0] def _evaluate_low_vol_overlay_target(self, macro: dict, park_mode: str) -> str | None: """Upgrade a safe parking regime to a stricter overlay symbol. @@ -5362,6 +6208,15 @@ class BacktestRunner: unrealized: float, ) -> DailyPortfolioState: gross_exposure, net_exposure = self._compute_portfolio_exposure(date) + # Idle capital decomposition: separate IA-sleeve vs primary notional + ia_exposure = 0.0 + for pos in self._open_positions: + if self._is_post_allocation_idle_engine_id(pos.plan.engine_id): + close = self._resolve_close_price( + pos.plan.candidate.symbol, date, pos.entry_price, + ) + ia_exposure += close * pos.shares_open + parking_val = self._get_parking_value(date) return DailyPortfolioState( date=date, equity=self._equity, @@ -5376,6 +6231,10 @@ class BacktestRunner: daily_new_risk_used=self._daily_new_risk_used, peak_equity=self._peak_equity, current_drawdown_pct=drawdown_pct, + raw_cash=self._cash, + parking_value=parking_val, + idle_alpha_exposure=ia_exposure, + primary_exposure=max(0.0, gross_exposure - ia_exposure), ) def _liquidate_parking(self, date: dt.date) -> bool: @@ -6511,7 +7370,7 @@ def main() -> None: parser.add_argument("--manifest", required=True, help="Path to experiment manifest JSON") parser.add_argument("--snapshot-id", help="Override dataset_snapshot_id") parser.add_argument("--snapshot-dir", help="Override snapshot root directory (default: data/parquet/)") - parser.add_argument("--split", default="train", help="Split name (train/valid/test)") + parser.add_argument("--split", default="train", help="Split name (train/valid/test/all). 'all' merges all splits for full-period backtest.") parser.add_argument("--output-root", default="./runs", help="Output root directory") parser.add_argument("--initial-equity", type=float, default=100_000.0) parser.add_argument("--config-root", default=".", help="Root dir for resolving config paths") @@ -6540,8 +7399,11 @@ def main() -> None: parser.add_argument("--end", default=None, help="End date filter YYYY-MM-DD (inclusive)") parser.add_argument("--parking", default=None, help="Cash parking preset (e.g. qqqm_low_dd)") parser.add_argument("--idle-alpha", default=None, help="Idle alpha sleeve preset (e.g. micro_event_alpha)") + parser.add_argument("--idle-alpha-dedup", default=None, choices=["skip", "rename"], help="IA dedup mode: 'skip' (default) skips conflicting IA engines; 'rename' adds __ia_sleeve suffix and injects anyway") parser.add_argument("--dividend-sleeve", default=None, help="Dividend capture sleeve preset (e.g. reserve_dividend_capture)") parser.add_argument("--form4-sleeve", default=None, help="Form 4 capture sleeve preset (e.g. reserve_form4_cluster)") + parser.add_argument("--ownership-sleeve", default=None, help="Ownership 13D/13G sleeve preset (e.g. ownership_13d_raise_reserve_plus_strict)") + parser.add_argument("--risk-off-sleeve", default=None, help="Risk-off alpha sleeve preset (e.g. risk_off_alpha_gld_dbc)") args = parser.parse_args() manifest = load_manifest(args.manifest) @@ -6554,6 +7416,8 @@ def main() -> None: config.risk.cash_parking_preset = args.parking config.risk.apply_parking_preset() if args.idle_alpha: + if args.idle_alpha_dedup: + config.idle_alpha_dedup_mode = args.idle_alpha_dedup config.idle_alpha_sleeve_preset = args.idle_alpha config.apply_idle_alpha_sleeve_preset() if args.dividend_sleeve: @@ -6562,6 +7426,12 @@ def main() -> None: if args.form4_sleeve: config.form4_capture_sleeve_preset = args.form4_sleeve config.apply_form4_capture_sleeve_preset() + if args.ownership_sleeve: + config.ownership_capture_sleeve_preset = args.ownership_sleeve + config.apply_ownership_capture_sleeve_preset() + if args.risk_off_sleeve: + config.risk_off_alpha_sleeve_preset = args.risk_off_sleeve + config.apply_risk_off_alpha_sleeve_preset() if args.walk_forward and args.robustness_matrix: raise SystemExit("Use either --walk-forward or --robustness-matrix, not both.") @@ -6593,7 +7463,10 @@ def main() -> None: end_date=dt.date.fromisoformat(args.end) if args.end else None, ) else: - store = _build_store(manifest, config, args.split, snapshot_dir_override=args.snapshot_dir) + if args.split == "all": + store = _build_merged_snapshot_store(manifest, config, snapshot_dir_override=args.snapshot_dir) + else: + store = _build_store(manifest, config, args.split, snapshot_dir_override=args.snapshot_dir) if args.start or args.end: all_store_dates = store.all_trading_days(include_reaction_dates=True) if not all_store_dates: @@ -6626,6 +7499,24 @@ def main() -> None: sqs_score, sqs_breakdown = compute_sqs(result.metrics) print(f"SQS: {sqs_score} ({', '.join(f'{k}={v}' for k, v in sqs_breakdown.items())})") + # Print sleeve decomposition if available + sleeve_path_str = result.artifact_paths.get("sleeve_decomposition") + sleeve_path = Path(sleeve_path_str) if sleeve_path_str else None + if sleeve_path and sleeve_path.exists(): + import json as _json + sd = _json.loads(sleeve_path.read_text()) + core_pct = sd.get("core", {}).get("contribution_pct", 0) + ia_pct = sd.get("idle_alpha", {}).get("contribution_pct", 0) + park_pct = sd.get("parking", {}).get("contribution_pct", 0) + amp = sd.get("composite_amplification") + idle = sd.get("avg_idle_fraction_pct") + amp_str = f" amp: {amp:.2f}x" if amp is not None else "" + idle_str = f" idle: {idle:.1f}%" if idle is not None else "" + print( + f"Sleeve: Core {core_pct:.1f}% | IA {ia_pct:.1f}% | Parking {park_pct:.1f}%" + f"{amp_str}{idle_str}" + ) + if __name__ == "__main__": main()