"""PEAD single-ticker advisor. `evaluate_pead_buy` mirrors the backtester's per-engine select_candidates loop on a single (ticker, asof). `evaluate_pead_sell` reconstructs the entry-day candidate to lock the engine-specific stop/target and runs the same `simulate_exit` the backtester uses against today's daily bar. """ from __future__ import annotations import datetime as dt import os from pathlib import Path from typing import Any, Literal from pydantic import BaseModel, Field from libs.backtest.allocator import ( _resolve_stop_risk_config, compute_stop_price, compute_target_price, ) from libs.backtest.domain import ( BacktestConfig, Candidate, OpenPosition, PlannedOrder, PositionStatus, ) from libs.backtest.execution import build_effective_execution_config, simulate_exit from libs.backtest.manifests import load_manifest, resolve_config from libs.backtest.selector import select_candidates from libs.common.logging import get_logger logger = get_logger(__name__) Signal = Literal["BUY", "SELL", "HOLD", "NO_SIGNAL", "ERROR"] class PeadVerdict(BaseModel): signal: Signal score: float | None = None reason: str details: dict[str, Any] = Field(default_factory=dict) warning: str | None = None def _oracle_url() -> str: return os.environ.get("ORACLE_URL") or os.environ.get( "STOCK_ORACLE_URL", "http://localhost:8000" ) def _db_dsn() -> str: return os.environ.get("DB_DSN") or os.environ.get("POSTGRES_DSN", "") def _load_config(config_path: str | Path) -> BacktestConfig: manifest = load_manifest(config_path) return resolve_config(manifest) async def _detector_rows_for_ticker( ticker: str, asof: dt.date, config: BacktestConfig, ) -> list[dict[str, Any]]: from apps.paper_trader.event_detector import EventDetector detector = EventDetector(db_dsn=_db_dsn(), oracle_url=_oracle_url()) rows = await detector.get_candidates_for_date(asof, config, convention="reaction_close") upper = ticker.upper() return [r for r in rows if str(r.get("symbol", "")).upper() == upper] def _try_each_engine( rows: list[dict[str, Any]], config: BacktestConfig, ) -> tuple[Candidate | None, str | None, list[str]]: """Loop enabled engines and return the first matching candidate. Returns (candidate, engine_id, engines_tried). When no engine matches, candidate and engine_id are None. """ engines = config.get_active_strategy_engines() tried: list[str] = [] for engine_cfg in engines: tried.append(engine_cfg.engine_id) candidates = select_candidates( raw_rows=rows, universe_config=config.universe, signal_config=config.signal, event_type_profiles=config.event_type_profiles or {}, strategy_engine=engine_cfg, ) if candidates: return candidates[0], engine_cfg.engine_id, tried return None, None, tried async def evaluate_pead_buy( ticker: str, asof: dt.date, config_path: str | Path, ) -> PeadVerdict: config = _load_config(config_path) rows = await _detector_rows_for_ticker(ticker, asof, config) if not rows: return PeadVerdict( signal="NO_SIGNAL", reason=( f"no qualifying filing event in pipeline DB for {ticker.upper()} on " f"{asof.isoformat()}; PEAD only acts on parsed filing events" ), ) candidate, engine_id, tried = _try_each_engine(rows, config) if candidate is None: return PeadVerdict( signal="NO_SIGNAL", reason=( f"event present for {ticker.upper()} on {asof.isoformat()} but no engine " f"accepted it (universe / score / event-type / direction gate failed for all " f"{len(tried)} engines)" ), details={"engines_tried": tried, "row_count": len(rows)}, ) return PeadVerdict( signal="BUY", score=candidate.score, reason=f"engine {engine_id} accepted; score={candidate.score:.3f}", details={ "engine_id": engine_id, "score": candidate.score, "event_type": candidate.event_type, "trade_direction": candidate.trade_direction, "entry_price_est": candidate.entry_price_est, "atr_14": candidate.atr_14, "sector": candidate.sector, "score_bucket": candidate.score_bucket, }, ) async def evaluate_pead_sell( ticker: str, entry_date: dt.date, entry_price: float, asof: dt.date, config_path: str | Path, ) -> PeadVerdict: config = _load_config(config_path) entry_rows = await _detector_rows_for_ticker(ticker, entry_date, config) if not entry_rows: return PeadVerdict( signal="ERROR", reason=( f"no pipeline-DB event for {ticker.upper()} on entry_date=" f"{entry_date.isoformat()}; cannot reconstruct stop/target — was this " "position taken under PEAD?" ), ) candidate, engine_id, tried = _try_each_engine(entry_rows, config) if candidate is None or engine_id is None: return PeadVerdict( signal="ERROR", reason=( f"event present on entry_date={entry_date.isoformat()} but no PEAD engine " f"would have accepted it (tried {len(tried)} engines); cannot reconstruct " "engine-specific stop/target deterministically" ), details={"engines_tried": tried}, ) exec_cfg = build_effective_execution_config(candidate, config) # Engine-locked stop distance based on entry-date ATR; translate to user's actual entry price. risk_cfg_for_stop = _resolve_stop_risk_config(candidate, config) synthetic_stop = compute_stop_price(candidate, risk_cfg_for_stop) risk_per_share = abs(candidate.entry_price_est - synthetic_stop) if risk_per_share <= 0: return PeadVerdict( signal="ERROR", reason="reconstructed stop distance is zero — atr_14 missing or invalid on entry_date", details={"engine_id": engine_id}, ) if candidate.trade_direction == "short": actual_stop = entry_price + risk_per_share else: actual_stop = max(0.01, entry_price - risk_per_share) target_r = ( candidate.engine_target_1_r if candidate.engine_target_1_r is not None else (exec_cfg.target_1_r if exec_cfg.target_1_r is not None else 2.0) ) actual_target = compute_target_price( entry_price_est=entry_price, stop_price=actual_stop, target_r=target_r, target_model=exec_cfg.target_model, target_atr_multiplier=exec_cfg.target_atr_multiplier, atr_14=candidate.atr_14, trade_direction=candidate.trade_direction, ) plan = PlannedOrder( candidate=candidate, shares=1, entry_price_limit=entry_price, stop_price=actual_stop, target_price=actual_target, risk_dollars=risk_per_share, engine_id=engine_id, entry_timing_policy=candidate.entry_timing_policy, ) days_held = _trading_day_count(entry_date, asof) position = OpenPosition( position_id=f"advisor_{ticker.upper()}_{entry_date.isoformat()}", plan=plan, entry_date=entry_date, entry_price=entry_price, entry_fill_slippage_bps=0.0, current_stop=actual_stop, target_price=actual_target, peak_price=entry_price, shares_open=1, shares_total=1, days_held=days_held, status=PositionStatus.ENTERED, ) bar = await _fetch_daily_bar(ticker, asof) base_details: dict[str, Any] = { "engine_id": engine_id, "stop_price": actual_stop, "target_price": actual_target, "days_held": days_held, "entry_date_atr_14": candidate.atr_14, "risk_per_share": risk_per_share, } warning = ( "stop/target reconstructed from entry-date ATR; trailing-stop and partial-exit " "state not modeled, so an actual backtest could exit earlier" ) if bar is None: return PeadVerdict( signal="HOLD", reason=( f"no daily bar available for {ticker.upper()} on {asof.isoformat()} " "(market may not have closed yet)" ), details=base_details, warning=warning, ) base_details.update({ "today_high": bar.get("high"), "today_low": bar.get("low"), "today_close": bar.get("close"), }) filled = simulate_exit(position, bar, exec_cfg, asof) if filled is None: return PeadVerdict( signal="HOLD", reason=f"no exit triggered today; days_held={days_held}", details=base_details, warning=warning, ) return PeadVerdict( signal="SELL", reason=f"exit triggered: {filled.exit_reason.value}", details={ **base_details, "exit_reason": filled.exit_reason.value, "exit_price": filled.exit_price, "pnl_pct": filled.pnl_pct, "r_multiple": filled.r_multiple, }, warning=warning, ) async def _fetch_daily_bar(ticker: str, asof: dt.date) -> dict[str, Any] | None: from libs.oracle_client.client import OracleClient from libs.oracle_client.price import PriceService async with OracleClient(base_url=_oracle_url()) as oc: svc = PriceService(oc) try: resp = await svc.get_daily_bars( ticker.upper(), start=asof.isoformat(), end=asof.isoformat() ) except Exception as exc: logger.warning("advisor_pead_bar_fetch_failed", ticker=ticker, asof=asof.isoformat(), error=str(exc)) return None for b in resp.bars: bar_date = b.date if isinstance(b.date, dt.date) else dt.date.fromisoformat(str(b.date)[:10]) if bar_date == asof: return {"open": b.open, "high": b.high, "low": b.low, "close": b.close, "volume": b.volume} if resp.bars: b = resp.bars[-1] return {"open": b.open, "high": b.high, "low": b.low, "close": b.close, "volume": b.volume} return None def _trading_day_count(entry_date: dt.date, asof: dt.date) -> int: if asof <= entry_date: return 0 try: from libs.backtest.calendar import get_trading_days days = get_trading_days(entry_date, asof) return max(0, len(days) - 1) except Exception: return (asof - entry_date).days