"""Chart Service（v3 §20.1）—— 只聚合与坐标整理，不重算研究结果。 - K 线/量/指标：基于主口径（adjust=none）行情，按请求 adjust 在**显示层**折算 qfq/hfq （绝不回写研究数据；研究执行仍用 price_adjustment 指定口径） - 标记：selections/signals 来自各自落库历史（by-symbol）；backtest fills 来自 Experiment 内 BacktestResult 的 trades/positions - 口径纪律（v3 §20.5）：显示价 basis 与回测执行价 basis 分别记录；不一致时对 marker 做 与 K 线相同的坐标换算，保证成交点贴图 """ from __future__ import annotations from datetime import date from app.domain.entities.chart import ( OHLC, ChartMetadata, ChartResult, EventMarker, SeriesPoint, VolumePoint, ) from app.domain.entities.market import AdjustFactor, DailyBar, Stock from app.domain.entities.research import BacktestResult from app.domain.repositories.market import ( AdjustFactorRepository, DailyBarRepository, StockRepository, ) _MA_WINDOWS = (20, 60) def _main_bars(bars: list[DailyBar]) -> list[DailyBar]: """仅取主口径行（adjust=none），丢弃新浪 qfq 兜底行 —— 显示层一律自 none 折算。""" return [b for b in bars if b.adjust == "none"] def _factor_multipliers( adj_repo: AdjustFactorRepository, symbol: str, start: date, end: date, mode: str, ) -> dict[date, float]: """返回 {trade_date: 显示折算系数}；mode=none → 空。 基准（v3 §20.5）：qfq 以**该股最新因子**（截至今天，而非图表区间末）归一， 保证历史区间随最新除权平移正确；hfq 直接用累积因子。 """ if mode == "none": return {} factors: list[AdjustFactor] = adj_repo.get_range(symbol, date(1990, 1, 1), date.today()) if not factors: return {} by_day = {f.trade_date: float(f.factor) for f in factors} latest = max(by_day.values()) out: dict[date, float] = {} for day, f in by_day.items(): out[day] = f / latest if mode == "qfq" else f return out class ChartService: def __init__( self, stock_repo: StockRepository, daily_repo: DailyBarRepository, adj_repo: AdjustFactorRepository, ) -> None: self._stock_repo = stock_repo self._daily_repo = daily_repo self._adj_repo = adj_repo def stock(self, symbol: str) -> Stock | None: return self._stock_repo.get_by_symbol(symbol) def stock_chart( self, symbol: str, start: date, end: date, adjust: str = "none", execution_price_basis: str | None = None, extra_markers: list[EventMarker] | None = None, ) -> ChartResult: """基础个股 K 线图（可叠加 fills 等外部标记）。""" stock = self.stock(symbol) name = stock.name if stock else "" raw = _main_bars(self._daily_repo.get_range(symbol, start, end)) mult = _factor_multipliers(self._adj_repo, symbol, start, end, adjust) bars: list[OHLC] = [] volume: list[VolumePoint] = [] for b in raw: m = mult.get(b.trade_date, 1.0) bars.append( OHLC( time=b.trade_date, open=_v(b.open, m), high=_v(b.high, m), low=_v(b.low, m), close=_v(b.close, m), ) ) volume.append(VolumePoint(time=b.trade_date, value=_v(b.volume, 1.0))) markers = _convert_markers(extra_markers or [], mult) indicators = _ma_indicators(bars) return ChartResult( metadata=ChartMetadata( symbol=symbol, name=name, adjust_mode=adjust, execution_price_basis=execution_price_basis, start=start, end=end, bar_count=len(bars), indicator_windows=list(_MA_WINDOWS), ), bars=bars, volume=volume, indicators=indicators, fills=[m for m in markers if m.kind.startswith("fill_")], signals=[m for m in markers if m.kind.startswith("signal_")], selections=[m for m in markers if m.kind == "selection"], holding_periods=_holding_periods(bars), ) # ---- 由已存历史构造标记（不重算） ---- def backtest_stock_chart( self, result: BacktestResult, symbol: str, start: date, end: date, adjust: str = "none", ) -> ChartResult: """回测个股视图：K 线 + 选股意图/未成交信号/实际成交三类标记（v3 §20.3）。""" basis = (result.config_snapshot or {}).get("price_adjustment", "none") markers = _result_to_markers(result, symbol) return self.stock_chart(symbol, start, end, adjust, execution_price_basis=basis, extra_markers=markers) def _result_to_markers(result: BacktestResult, symbol: str) -> list[EventMarker]: """由回测 history 生成个股标记：fills（成交）/ signals（未成交意图）/ selections（选股）。""" markers: list[EventMarker] = [] # 实际成交（fills）与未成交信号（signal_history 中 filled=False） for a in result.signal_history: if a.symbol != symbol: continue if a.filled: kind = "fill_buy" if a.signal == "BUY" else "fill_sell" text = [f"{'买入' if a.signal=='BUY' else '卖出'} @ {a.price:.2f}（basis={_basis_of(result)}）"] markers.append( EventMarker(time=a.date, kind=kind, symbol=symbol, price=a.price, text=text) ) else: kind = "signal_buy" if a.signal == "BUY" else "signal_sell" text = [a.reject_reason or f"{a.signal} 未成交"] markers.append(EventMarker(time=a.date, kind=kind, symbol=symbol, price=a.price, text=text)) # 选股意图（selection_history 中该 symbol 的命中） for pk in result.selection_history: if pk.symbol != symbol: continue markers.append( EventMarker( time=pk.date, kind="selection", symbol=symbol, score=pk.score, text=[f"选股意图 rank #{pk.rank}"], ) ) return markers def _basis_of(result: BacktestResult) -> str: return (result.config_snapshot or {}).get("price_adjustment", "none") def _convert_markers(markers: list[EventMarker], mult: dict[date, float]) -> list[EventMarker]: """显示口径与执行价 basis 不一致时，把 marker 价格折算到 K 线坐标系（v3 §20.5）。""" out: list[EventMarker] = [] for m in markers: if m.price is not None and mult: k = mult.get(m.time) if k is not None: m = m.model_copy(update={"price": round(m.price * k, 4)}) out.append(m) return out def _holding_periods(bars: list[OHLC]) -> list[dict]: """v1 空实现占位（持仓区间渲染 v3 §20.4 后续细化）。""" return [] def _ma_indicators(bars: list[OHLC]) -> dict[str, list[SeriesPoint]]: import statistics closes = [b.close for b in bars] out: dict[str, list[SeriesPoint]] = {} for w in _MA_WINDOWS: series: list[SeriesPoint] = [] for i, b in enumerate(bars): if i + 1 < w: continue window = closes[i + 1 - w : i + 1] if all(v is not None for v in window): series.append(SeriesPoint(time=b.time, value=round(statistics.fmean(window), 4))) out[f"ma{w}"] = series return out def _v(v, m: float) -> float | None: if v is None: return None f = float(v) return round(f * m, 4)