Simon
995ed08548
- domain/entities/chart.py：ChartResult/OHLC/Volume/Series/EventMarker/ChartMetadata
（adjust_mode + execution_price_basis 口径元数据）+ SelectionHit
- application/services/chart_service.py：个股 K线/量/MA 指标；显示层 qfq/hfq 折算
（基于主口径 none 行情 × adjust_factor，绝回写研究数据）；回测个股视图把实际成交
转 fills 标记并在显示口径不同时做坐标换算（v3 §20.3/§20.5）
- by-symbol 历史查询：SignalRepository/SelectionRepository.list_by_symbol（含溯源 id）
- api/charts.py：/stocks/{symbol}/chart|signals|selections、/backtests/{id}/stocks/{symbol}/chart
|trades|positions
- tests/test_charts.py（指标/qfq-hfq 折算断言/回测 fills/API 集成+404）；全量 pytest 通过
207 lines
6.9 KiB
Python
207 lines
6.9 KiB
Python
"""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, Trade
|
||
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 → 空。qfq: f/f_latest；hfq: f。"""
|
||
if mode == "none":
|
||
return {}
|
||
factors: list[AdjustFactor] = adj_repo.get_range(symbol, date(1990, 1, 1), end)
|
||
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 线 + 该股实际成交 fills（v3 §20.3 Signal↔Fill 展示）。"""
|
||
basis = (result.config_snapshot or {}).get("price_adjustment", "none")
|
||
markers = _trades_to_markers(result.trades, symbol, basis)
|
||
return self.stock_chart(symbol, start, end, adjust, execution_price_basis=basis,
|
||
extra_markers=markers)
|
||
|
||
|
||
def _trades_to_markers(trades: list[Trade], symbol: str, basis: str) -> list[EventMarker]:
|
||
markers: list[EventMarker] = []
|
||
for t in trades:
|
||
if t.symbol != symbol:
|
||
continue
|
||
markers.append(
|
||
EventMarker(
|
||
time=t.entry_date,
|
||
kind="fill_buy",
|
||
symbol=symbol,
|
||
price=t.entry_price,
|
||
text=[f"买入 @ {t.entry_price:.2f}（basis={basis}）"],
|
||
)
|
||
)
|
||
markers.append(
|
||
EventMarker(
|
||
time=t.exit_date,
|
||
kind="fill_sell",
|
||
symbol=symbol,
|
||
price=t.exit_price,
|
||
text=[f"卖出 @ {t.exit_price:.2f}，收益 {t.return_pct:.2f}%（basis={basis}）"],
|
||
)
|
||
)
|
||
return markers
|
||
|
||
|
||
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)
|