feat(selection): implement gold brick resonance strategy with evaluation and logging enhancements

This commit is contained in:
yuxuanhui
2026-09-05 10:22:08 +08:00
parent 9163590070
commit 79476252b1
20 changed files with 1060 additions and 97 deletions
@@ -0,0 +1,59 @@
"""Application use case for one-stock historical gold-brick evaluation."""
from __future__ import annotations
from collections.abc import Sequence
from datetime import date
from ..domain.gold_brick import GoldBrickStrategy
from ..domain.models import SelectionEvaluation, StockHistory
from ..domain.ports import MarketDataReader, MarketDataReaderError
class EvaluateGoldBrick:
"""Read one history, evaluate gold-brick, and map read failures."""
def __init__(
self,
reader: MarketDataReader,
strategy: GoldBrickStrategy | None = None,
) -> None:
"""Inject the market-data port and optionally a strategy instance."""
self.reader = reader
self.strategy = strategy or GoldBrickStrategy()
def execute(self, ts_code: str, target_trade_date: date) -> SelectionEvaluation:
"""Evaluate ``ts_code`` on the exact requested trading date."""
try:
history = self.reader.load_history(ts_code, target_trade_date)
except MarketDataReaderError as exc:
return SelectionEvaluation(
ts_code=ts_code,
target_trade_date=target_trade_date,
status="data_error",
reason=str(exc),
)
return self.strategy.evaluate(history, target_trade_date)
def execute_history(
self,
history: StockHistory,
target_trade_date: date,
) -> SelectionEvaluation:
"""Evaluate a history already loaded by the bounded batch reader."""
return self.strategy.evaluate(history, target_trade_date)
def execute_histories(
self,
histories: Sequence[StockHistory],
target_trade_date: date,
) -> tuple[SelectionEvaluation, ...]:
"""Evaluate loaded histories without issuing one read per stock."""
return tuple(self.execute_history(history, target_trade_date) for history in histories)
__all__ = ["EvaluateGoldBrick"]
@@ -1,16 +1,17 @@
"""Application orchestration for persisted whole-universe B1 runs."""
"""Application orchestration for persisted whole-universe strategy runs."""
from __future__ import annotations
import logging
import time
from collections.abc import Callable, Sequence
from collections import Counter
from collections.abc import Callable, Mapping, Sequence
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import date
from typing import Literal, Protocol, cast
from typing import Protocol, cast
from ..domain.models import SelectionEvaluation, StockHistory
from ..domain.models import SelectionEvaluation, SelectionStrategyName, StockHistory
from ..domain.pattern_scoring import (
PatternCase,
PatternCaseLibraryLoader,
@@ -33,9 +34,16 @@ from ..domain.runs import (
)
from .evaluate import EvaluateZhixingB1
logger = logging.getLogger(__name__)
StrategyName = Literal["zhixing_b1"]
_FAILURE_STATUSES = {"insufficient_history", "missing_target_bar", "data_error"}
# Selection runs are started by the ASGI service in production. A child of
# Uvicorn's configured logger keeps INFO diagnostics visible in container logs.
logger = logging.getLogger("uvicorn.error.zhixing.selection.run")
StrategyName = SelectionStrategyName
_FAILURE_STATUSES = {
"insufficient_history",
"missing_target_bar",
"missing_turnover_rate",
"data_error",
}
class SelectionEvaluator(Protocol):
@@ -53,13 +61,18 @@ class PreparedSelectionRun:
class RunZhixingB1:
"""Prepare, execute, and query persisted Zhixing B1 result batches."""
"""Prepare, execute, and query persisted selection strategy batches.
The historical class name remains as a compatibility seam for existing
composition and tests while strategy routing is now explicit.
"""
def __init__(
self,
reader: SelectionUniverseReader,
store: SelectionRunStore,
evaluator: SelectionEvaluator | None = None,
evaluators: Mapping[StrategyName, SelectionEvaluator] | None = None,
pattern_case_loader: PatternCaseLibraryLoader | None = None,
pattern_scorer: PatternScorer | None = None,
*,
@@ -76,6 +89,11 @@ class RunZhixingB1:
self.reader = reader
self.store = store
self.evaluator = evaluator or EvaluateZhixingB1(reader)
self.evaluators: dict[StrategyName, SelectionEvaluator] = {
"zhixing_b1": self.evaluator,
}
if evaluators is not None:
self.evaluators.update(evaluators)
self.pattern_case_loader = pattern_case_loader
self.pattern_scorer = pattern_scorer
self.pattern_scoring_enabled = pattern_scoring_enabled
@@ -91,12 +109,43 @@ class RunZhixingB1:
) -> PreparedSelectionRun:
"""Validate source eligibility before claiming the rerunnable key."""
source = self.reader.load_execution_source(strategy, target_trade_date)
run = self.store.prepare_run(
logger.info(
"selection_run_prepare_started strategy=%s target_trade_date=%s rerun=%s",
strategy,
target_trade_date,
source,
rerun=rerun,
target_trade_date.isoformat(),
rerun,
)
if strategy not in self.evaluators:
raise ValueError(f"selection evaluator is not configured for strategy: {strategy}")
try:
source = self.reader.load_execution_source(strategy, target_trade_date)
run = self.store.prepare_run(
strategy,
target_trade_date,
source,
rerun=rerun,
)
except Exception as exc: # noqa: BLE001 - log the safe prepare boundary and preserve type
logger.warning(
"selection_run_prepare_failed strategy=%s target_trade_date=%s "
"status=failed error_type=%s reason=%s",
strategy,
target_trade_date.isoformat(),
exc.__class__.__name__,
_safe_item_error(exc),
)
raise
logger.info(
"selection_run_prepared strategy=%s target_trade_date=%s run_id=%s "
"market_sync_batch_id=%s target_count=%d eligible_count=%d "
"coverage=%s status=running",
strategy,
target_trade_date.isoformat(),
run.id,
source.market_sync_batch_id,
source.target_count,
len(source.stocks),
source.coverage,
)
return PreparedSelectionRun(run=run, source=source)
@@ -109,29 +158,76 @@ class RunZhixingB1:
"""
stocks = _unique_stocks(prepared.source.stocks)
strategy = prepared.run.strategy
target_trade_date = prepared.source.target_trade_date
evaluator = self.evaluators[strategy]
evaluated_count = 0
selected_stock_count = 0
signal_count = 0
failed_count = 0
missing_turnover_count = 0
insufficient_history_count = 0
history_rows = 0
batch_count = _chunk_count(len(stocks), self.batch_size)
current_batch = 0
final_status: SelectionRunStatus = "failed"
read_seconds = 0.0
evaluate_seconds = 0.0
persist_seconds = 0.0
scoring_seconds = 0.0
logger.info(
"selection_run_started strategy=%s target_trade_date=%s run_id=%s "
"market_sync_batch_id=%s stock_count=%d batch_count=%d worker_count=%d "
"status=running",
strategy,
target_trade_date.isoformat(),
prepared.run.id,
prepared.source.market_sync_batch_id,
len(stocks),
batch_count,
self.max_workers,
)
try:
pattern_cases, pattern_library_error = self._prepare_pattern_cases(prepared.run.id)
if strategy == "zhixing_b1":
pattern_cases, pattern_library_error = self._prepare_pattern_cases(prepared.run.id)
else:
pattern_cases, pattern_library_error = None, None
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
for batch_stocks in _chunks(stocks, self.batch_size):
for batch_index, batch_stocks in enumerate(
_chunks(stocks, self.batch_size),
start=1,
):
current_batch = batch_index
read_started = time.perf_counter()
histories = self._load_histories(
batch_stocks,
prepared.source.target_trade_date,
target_trade_date,
evaluator,
)
read_seconds += time.perf_counter() - read_started
history_rows += sum(
batch_read_seconds = time.perf_counter() - read_started
read_seconds += batch_read_seconds
batch_history_rows = sum(
len(history.bars) for history in histories if history is not None
)
history_rows += batch_history_rows
batch_missing_turnover = sum(
not _turnover_present(history, target_trade_date)
for history in histories
)
logger.info(
"selection_read_batch_summary strategy=%s target_trade_date=%s "
"run_id=%s batch=%d batch_count=%d stock_count=%d history_rows=%d "
"turnover_missing_count=%d status=success read_seconds=%.3f",
strategy,
target_trade_date.isoformat(),
prepared.run.id,
batch_index,
batch_count,
len(batch_stocks),
batch_history_rows,
batch_missing_turnover,
batch_read_seconds,
)
evaluate_started = time.perf_counter()
evaluations = tuple(
@@ -139,10 +235,46 @@ class RunZhixingB1:
self._evaluate_stock,
batch_stocks,
histories,
[prepared.source.target_trade_date] * len(batch_stocks),
[target_trade_date] * len(batch_stocks),
[evaluator] * len(batch_stocks),
[strategy] * len(batch_stocks),
[prepared.run.id] * len(batch_stocks),
[batch_index] * len(batch_stocks),
)
)
evaluate_seconds += time.perf_counter() - evaluate_started
batch_evaluate_seconds = time.perf_counter() - evaluate_started
evaluate_seconds += batch_evaluate_seconds
status_counts = Counter(evaluation.status for evaluation in evaluations)
no_signal_reasons = Counter(
evaluation.reason or "unspecified"
for evaluation in evaluations
if evaluation.status == "no_signal"
)
missing_turnover_count += status_counts["missing_turnover_rate"]
insufficient_history_count += status_counts["insufficient_history"]
for stock, history, evaluation in zip(
batch_stocks,
histories,
evaluations,
strict=True,
):
if evaluation.status not in _FAILURE_STATUSES:
continue
logger.warning(
"selection_item_incomplete strategy=%s target_trade_date=%s "
"run_id=%s batch=%d ts_code=%s history_rows=%d "
"turnover_present=%s status=%s error_type=%s reason=%s",
strategy,
target_trade_date.isoformat(),
prepared.run.id,
batch_index,
stock.ts_code,
len(history.bars) if history is not None else 0,
_turnover_present(history, target_trade_date),
evaluation.status,
evaluation.status,
evaluation.reason or evaluation.status,
)
scoring_started = time.perf_counter()
items = tuple(
@@ -151,6 +283,7 @@ class RunZhixingB1:
stock.name,
evaluation,
pattern_score=self._score_stock(
strategy,
prepared.run.id,
stock,
history,
@@ -173,23 +306,79 @@ class RunZhixingB1:
signal_count += sum(item.signal_count for item in items)
failed_count += sum(item.status in _FAILURE_STATUSES for item in items)
logger.info(
"selection_evaluate_batch_summary strategy=%s target_trade_date=%s "
"run_id=%s batch=%d batch_count=%d stock_count=%d selected_count=%d "
"no_signal_count=%d insufficient_history_count=%d "
"missing_target_bar_count=%d missing_turnover_count=%d "
"data_error_count=%d no_signal_reasons=%s status=complete "
"evaluate_seconds=%.3f",
strategy,
target_trade_date.isoformat(),
prepared.run.id,
batch_index,
batch_count,
len(items),
status_counts["selected"],
status_counts["no_signal"],
status_counts["insufficient_history"],
status_counts["missing_target_bar"],
status_counts["missing_turnover_rate"],
status_counts["data_error"],
dict(no_signal_reasons),
batch_evaluate_seconds,
)
persist_started = time.perf_counter()
self._record_items(prepared.run.id, items)
persist_seconds += time.perf_counter() - persist_started
batch_persist_seconds = time.perf_counter() - persist_started
persist_seconds += batch_persist_seconds
logger.info(
"selection_persist_batch_summary strategy=%s target_trade_date=%s "
"run_id=%s batch=%d batch_count=%d item_count=%d signal_count=%d "
"status=success persist_seconds=%.3f",
strategy,
target_trade_date.isoformat(),
prepared.run.id,
batch_index,
batch_count,
len(items),
sum(item.signal_count for item in items),
batch_persist_seconds,
)
status = _run_status(evaluated_count, failed_count)
final_status = _run_status(evaluated_count, failed_count)
self.store.finish_run(
prepared.run.id,
status,
final_status,
evaluated_count=evaluated_count,
selected_stock_count=selected_stock_count,
signal_count=signal_count,
failed_count=failed_count,
)
except Exception as exc: # noqa: BLE001 - worker boundary must persist failure state
logger.error(
"selection_run_failed run_id=%s error_type=%s reason=%s",
logger.info(
"selection_run_converged strategy=%s target_trade_date=%s run_id=%s "
"batch=%d evaluated_count=%d selected_stock_count=%d signal_count=%d "
"failed_count=%d status=%s",
strategy,
target_trade_date.isoformat(),
prepared.run.id,
current_batch,
evaluated_count,
selected_stock_count,
signal_count,
failed_count,
final_status,
)
except Exception as exc: # noqa: BLE001 - worker boundary must persist failure state
final_status = "failed"
logger.error(
"selection_run_failed strategy=%s target_trade_date=%s run_id=%s "
"batch=%d status=failed error_type=%s reason=%s",
strategy,
target_trade_date.isoformat(),
prepared.run.id,
current_batch,
exc.__class__.__name__,
_safe_item_error(exc),
)
@@ -202,23 +391,41 @@ class RunZhixingB1:
signal_count=signal_count,
failed_count=max(failed_count, 1),
error_type="batch_error",
error_message=str(exc),
error_message=_safe_item_error(exc),
)
except Exception: # noqa: BLE001 - preserve the original worker failure
logger.error(
"selection_run_failure_persist_failed run_id=%s",
"selection_run_failure_persist_failed strategy=%s target_trade_date=%s "
"run_id=%s batch=%d status=failed error_type=finish_run_failed",
strategy,
target_trade_date.isoformat(),
prepared.run.id,
current_batch,
)
finally:
logger.info(
"selection_run_summary run_id=%s stock_count=%d history_rows=%d "
"batch_count=%d worker_count=%d read_seconds=%.3f "
"selection_run_summary strategy=%s target_trade_date=%s run_id=%s "
"market_sync_batch_id=%s stock_count=%d history_rows=%d batch_count=%d "
"last_batch=%d worker_count=%d evaluated_count=%d selected_stock_count=%d "
"signal_count=%d failed_count=%d insufficient_history_count=%d "
"missing_turnover_count=%d status=%s read_seconds=%.3f "
"evaluate_seconds=%.3f scoring_seconds=%.3f persist_seconds=%.3f",
strategy,
target_trade_date.isoformat(),
prepared.run.id,
prepared.source.market_sync_batch_id,
len(stocks),
history_rows,
batch_count,
current_batch,
self.max_workers,
evaluated_count,
selected_stock_count,
signal_count,
failed_count,
insufficient_history_count,
missing_turnover_count,
final_status,
read_seconds,
evaluate_seconds,
scoring_seconds,
@@ -251,6 +458,7 @@ class RunZhixingB1:
def _score_stock(
self,
strategy: StrategyName,
run_id: str,
stock: SelectionStock,
history: StockHistory | None,
@@ -260,7 +468,11 @@ class RunZhixingB1:
) -> PatternScore:
"""Score one selected stock once and isolate enrichment failures."""
if not self.pattern_scoring_enabled or evaluation.status != "selected":
if (
strategy != "zhixing_b1"
or not self.pattern_scoring_enabled
or evaluation.status != "selected"
):
return PatternScore()
if library_error is not None:
return PatternScore.failed(library_error)
@@ -283,6 +495,7 @@ class RunZhixingB1:
self,
stocks: Sequence[SelectionStock],
target_trade_date: date,
evaluator: SelectionEvaluator,
) -> tuple[StockHistory | None, ...]:
"""Load one chunk when the reader supports it, with old-path fallback."""
@@ -300,7 +513,8 @@ class RunZhixingB1:
for stock in typed_stocks
)
if isinstance(self.evaluator, EvaluateZhixingB1):
execute_history = getattr(evaluator, "execute_history", None)
if callable(execute_history):
return tuple(
self.reader.load_history(stock.ts_code, target_trade_date) for stock in typed_stocks
)
@@ -311,23 +525,35 @@ class RunZhixingB1:
stock: SelectionStock,
history: StockHistory | None,
target_trade_date: date,
evaluator: SelectionEvaluator,
strategy: StrategyName,
run_id: str,
batch_index: int,
) -> SelectionEvaluation:
"""Evaluate one stock inside a worker and isolate its exception."""
ts_code = stock.ts_code
try:
execute_history: Callable[[StockHistory, date], SelectionEvaluation] | None = getattr(
self.evaluator,
evaluator,
"execute_history",
None,
)
if history is not None and execute_history is not None:
return execute_history(history, target_trade_date)
return self.evaluator.execute(ts_code, target_trade_date)
return evaluator.execute(ts_code, target_trade_date)
except Exception as exc: # noqa: BLE001 - isolate one stock from the batch
logger.warning(
"selection_item_failed ts_code=%s error_type=%s reason=%s",
"selection_item_failed strategy=%s target_trade_date=%s run_id=%s "
"batch=%d ts_code=%s history_rows=%d turnover_present=%s "
"status=data_error error_type=%s reason=%s",
strategy,
target_trade_date.isoformat(),
run_id,
batch_index,
ts_code,
len(history.bars) if history is not None else 0,
_turnover_present(history, target_trade_date),
exc.__class__.__name__,
_safe_item_error(exc),
)
@@ -407,6 +633,15 @@ def _safe_item_error(error: Exception) -> str:
return " ".join(str(error).split())[:500] or error.__class__.__name__
def _turnover_present(history: StockHistory | None, target_trade_date: date) -> bool:
"""Return whether target-day Tushare turnover is available for diagnostics."""
if history is None:
return False
basic = history.daily_basic.get(target_trade_date)
return basic is not None and basic.turnover_rate is not None
def _chunks(
values: Sequence[SelectionStock],
size: int,
@@ -0,0 +1,381 @@
"""Formula-level implementation of the independent gold-brick strategy."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date
import numpy as np
import pandas as pd
from .indicators import EXIST, HHV, LLV, REF, SMA, serializable_metrics
from .models import (
GoldBrickCategory,
SelectionEvaluation,
SelectionSignal,
StockHistory,
)
from .zhixing_b1 import compute_signal_masks, prepare_zhixing_b1_indicators
GOLD_BRICK_MINIMUM_HISTORY = 200
GOLD_BRICK_TURNOVER_RATE_THRESHOLD = 0.99
GOLD_BRICK_SIGNAL_ORDER: tuple[GoldBrickCategory, ...] = (
GoldBrickCategory.RESONANCE,
)
def _safe_ratio(numerator: pd.Series, denominator: pd.Series) -> pd.Series:
"""Divide two series while retaining invalid zero denominators as NaN."""
return numerator.div(denominator.replace(0, np.nan))
def prepare_gold_brick_indicators(frame: pd.DataFrame, code: str) -> pd.DataFrame:
"""Prepare the original gold-brick formula on ascending qfq OHLCV rows.
Args:
frame: Ascending qfq rows with ``open``, ``high``, ``low``, ``close``
and ``volume`` columns.
code: Tushare-style stock code used by the reused B1 width rules.
Returns:
A prepared frame containing the seven B1 masks, brick chart, momentum,
trend, upper-shadow, and both resonance conditions.
"""
result = prepare_zhixing_b1_indicators(frame, code)
close = result["close"].astype(float)
high = result["high"].astype(float)
low = result["low"].astype(float)
open_price = result["open"].astype(float)
volume = result["volume"].astype(float)
previous_close = REF(close, 1)
previous_volume = REF(volume, 1)
range_high = HHV(high, 4)
range_low = LLV(low, 4)
range_width = range_high - range_low
var1a = _safe_ratio(range_high - close, range_width).mul(100).sub(90)
var2a = SMA(var1a, 4, 1).add(100)
var3a = _safe_ratio(close - range_low, range_width).mul(100)
var4a = SMA(var3a, 6, 1)
var5a = SMA(var4a, 6, 1).add(100)
var6a = var5a - var2a
result["brick_chart"] = (var6a - 4).where(var6a > 4, 0.0)
b1_masks = compute_signal_masks(result)
existing_b1 = pd.Series(False, index=result.index, dtype=bool)
for mask in b1_masks.values():
existing_b1 |= mask
result["existing_b1"] = existing_b1
j_momentum = result["j"] - REF(result["j"], 1)
rsi_momentum = result["rsi"] - REF(result["rsi"], 1)
momentum_sum = j_momentum + rsi_momentum
previous_momentum_sum = REF(j_momentum, 1) + REF(rsi_momentum, 1)
volume_ratio = _safe_ratio(volume, previous_volume)
volume_coefficient = pd.Series(
np.where(
volume < previous_volume * 0.99,
(1 - 5 * _safe_ratio(previous_volume - volume, previous_volume)) * 0.8,
1.0,
),
index=result.index,
dtype=float,
)
multiple_volume_coefficient = pd.Series(
np.where(volume_ratio >= 4, 1.4, volume_ratio * 0.1 + 1),
index=result.index,
dtype=float,
)
multiple_volume_bonus = pd.Series(
np.where(
(close > open_price)
& (close > previous_close)
& (volume > previous_volume * 1.8),
multiple_volume_coefficient,
1.0,
),
index=result.index,
dtype=float,
)
shadow_floor = pd.Series(
np.minimum(open_price.to_numpy(float), previous_close.to_numpy(float)),
index=result.index,
dtype=float,
)
shadow_coefficient = pd.Series(
np.where(
(close > previous_close) & (close > open_price),
(0.75 - _safe_ratio(high - close, high - shadow_floor)) * 1.3,
1.0,
),
index=result.index,
dtype=float,
)
result["j_momentum"] = j_momentum
result["rsi_momentum"] = rsi_momentum
result["yellow_column"] = (
momentum_sum.div(2).mul(shadow_coefficient).mul(multiple_volume_bonus)
)
x_condition = (
(close > open_price)
& (close > previous_close)
& (momentum_sum > previous_momentum_sum)
)
result["x_momentum"] = (
momentum_sum.sub(previous_momentum_sum)
.div(2)
.mul(shadow_coefficient)
.mul(volume_coefficient)
.mul(multiple_volume_bonus)
.where(x_condition, 0.0)
)
brick = result["brick_chart"]
current_red = brick > REF(brick, 1)
current_green = brick <= REF(brick, 1)
previous_green = REF(current_green.astype(float), 1) == 1
red_length = (brick - REF(brick, 1)).where(current_red, 0.0)
brick_length = brick - REF(brick, 1)
previous_green_length = (REF(brick, 2) - REF(brick, 1)).where(
previous_green,
0.0,
)
red_green_ratio = _safe_ratio(red_length, previous_green_length).where(
previous_green_length > 0,
0.0,
)
result["brick_length"] = brick_length
result["strong_red"] = current_red & previous_green & (red_green_ratio > 0.666)
result["gold_trend_condition"] = (
(result["trend_white"] >= result["trend_yellow"] * 0.995)
& (result["trend_yellow"] >= REF(result["trend_yellow"], 1) * 0.997)
& (close >= result["trend_yellow"] * 0.997)
)
upper_shadow_floor = pd.Series(
np.minimum(low.to_numpy(float), previous_close.to_numpy(float)),
index=result.index,
dtype=float,
)
result["upper_shadow_strength"] = 1 - _safe_ratio(
high - close,
high - upper_shadow_floor,
)
result["upper_shadow_condition"] = (
((close >= open_price) | (close > previous_close))
& (result["upper_shadow_strength"] > 0.618)
)
long = result["long_oscillator"]
short = result["short_oscillator"]
result["resonance_condition_1"] = (
result["strong_red"]
& ((result["yellow_column"] >= 7.5) | (result["x_momentum"] >= 7.5))
& (EXIST(existing_b1, 2) | ((REF(long, 1) > 85) & (REF(short, 1) < 30)))
)
result["resonance_condition_2"] = (
result["strong_red"]
& ((result["yellow_column"] >= 10) | (result["x_momentum"] >= 10))
& (
(EXIST((long - short) > 60, 4) & (long > 98) & (short > 98))
| ((result["yellow_column"] > 20) & (close > result["trend_white"]))
| (result["yellow_column"] > 30)
| ((result["yellow_column"] + brick_length) > 50)
| (result["x_momentum"] > 40)
)
)
result["resonance_condition"] = (
result["resonance_condition_1"] | result["resonance_condition_2"]
)
return result
@dataclass(frozen=True, slots=True)
class GoldBrickStrategy:
"""Evaluate the close-of-day gold-brick resonance formula."""
name: str = "gold_brick"
def evaluate(
self,
history: StockHistory,
target_trade_date: date,
) -> SelectionEvaluation:
"""Evaluate one explicit date and fail closed on incomplete formula inputs."""
bars_by_date = {bar.trade_date: bar for bar in history.bars}
target_bar = bars_by_date.get(target_trade_date)
if target_bar is None or any(
value is None
for value in (
target_bar.open,
target_bar.high,
target_bar.low,
target_bar.close,
target_bar.volume,
)
):
return SelectionEvaluation(
history.ts_code,
target_trade_date,
"missing_target_bar",
reason="target trade date has no complete qfq daily bar",
)
selected_bars = tuple(
sorted(
(bar for bar in bars_by_date.values() if bar.trade_date <= target_trade_date),
key=lambda bar: bar.trade_date,
)
)
if len(selected_bars) < GOLD_BRICK_MINIMUM_HISTORY:
return SelectionEvaluation(
history.ts_code,
target_trade_date,
"insufficient_history",
reason=(
f"gold brick needs at least {GOLD_BRICK_MINIMUM_HISTORY} "
"ascending bars before evaluation"
),
)
if any(
value is None
for bar in selected_bars
for value in (bar.open, bar.high, bar.low, bar.close, bar.volume)
):
return SelectionEvaluation(
history.ts_code,
target_trade_date,
"data_error",
reason="gold brick history contains incomplete qfq OHLCV values",
)
target_basic = history.daily_basic.get(target_trade_date)
turnover_rate = target_basic.turnover_rate if target_basic is not None else None
if turnover_rate is None:
return SelectionEvaluation(
history.ts_code,
target_trade_date,
"missing_turnover_rate",
reason="target trade date has no Tushare turnover_rate",
)
frame = pd.DataFrame(
{
"trade_date": [bar.trade_date for bar in selected_bars],
"open": [bar.open for bar in selected_bars],
"high": [bar.high for bar in selected_bars],
"low": [bar.low for bar in selected_bars],
"close": [bar.close for bar in selected_bars],
"volume": [bar.volume for bar in selected_bars],
}
)
prepared = prepare_gold_brick_indicators(frame, history.ts_code)
target_index = int(prepared.index[prepared["trade_date"] == target_trade_date][0])
required_metrics = (
"brick_chart",
"brick_length",
"yellow_column",
"x_momentum",
"trend_white",
"trend_yellow",
"upper_shadow_strength",
)
invalid_metrics = tuple(
metric
for metric in required_metrics
if not np.isfinite(float(prepared.at[target_index, metric]))
)
if invalid_metrics:
return SelectionEvaluation(
history.ts_code,
target_trade_date,
"data_error",
reason=(
"gold brick formula produced non-finite target metrics: "
+ ", ".join(invalid_metrics)
),
)
row = prepared.iloc[target_index]
turnover_condition = turnover_rate >= GOLD_BRICK_TURNOVER_RATE_THRESHOLD
matched = bool(
row["resonance_condition"]
and row["upper_shadow_condition"]
and row["gold_trend_condition"]
and turnover_condition
)
if not matched:
gate_state = (
f"resonance={int(bool(row['resonance_condition']))} "
f"upper_shadow={int(bool(row['upper_shadow_condition']))} "
f"trend={int(bool(row['gold_trend_condition']))} "
f"turnover={int(turnover_condition)}"
)
return SelectionEvaluation(
history.ts_code,
target_trade_date,
"no_signal",
reason=f"gold brick gates did not match: {gate_state}",
)
resonance_types = ";".join(
label
for label, condition in (
("共振条件1", row["resonance_condition_1"]),
("共振条件2", row["resonance_condition_2"]),
)
if bool(condition)
)
details = serializable_metrics(
(
("signal", "金砖共振"),
("resonance_type", resonance_types),
("brick_chart", row["brick_chart"]),
("brick_length", row["brick_length"]),
("yellow_column", row["yellow_column"]),
("x_momentum", row["x_momentum"]),
("j", row["j"]),
("rsi", row["rsi"]),
("trend_white", row["trend_white"]),
("trend_yellow", row["trend_yellow"]),
("upper_shadow_strength", row["upper_shadow_strength"]),
("turnover_rate", turnover_rate),
)
)
signal = SelectionSignal(
ts_code=history.ts_code,
name=history.name,
target_trade_date=target_trade_date,
strategy="gold_brick",
category=GoldBrickCategory.RESONANCE,
close=float(row["close"]),
details=details,
)
return SelectionEvaluation(
history.ts_code,
target_trade_date,
"selected",
signals=(signal,),
)
def select(
self,
history: StockHistory,
target_trade_date: date,
) -> tuple[SelectionSignal, ...]:
"""Return only signals for callers that do not need evaluation status."""
return self.evaluate(history, target_trade_date).signals
__all__ = [
"GOLD_BRICK_MINIMUM_HISTORY",
"GOLD_BRICK_SIGNAL_ORDER",
"GOLD_BRICK_TURNOVER_RATE_THRESHOLD",
"GoldBrickStrategy",
"prepare_gold_brick_indicators",
]
@@ -9,6 +9,8 @@ from enum import StrEnum
from math import isfinite
from typing import Literal
SelectionStrategyName = Literal["zhixing_b1", "gold_brick"]
def _validate_number(value: float | None, field_name: str) -> None:
"""Reject infinities while allowing ``None`` for incomplete source rows."""
@@ -91,15 +93,24 @@ class ZhixingB1Category(StrEnum):
PULLBACK_YELLOW = "zhixing_b1_pullback_yellow"
class GoldBrickCategory(StrEnum):
"""The independently persisted final signal from the gold-brick formula."""
RESONANCE = "gold_brick_resonance"
SelectionSignalCategory = ZhixingB1Category | GoldBrickCategory
@dataclass(frozen=True, slots=True)
class SelectionSignal:
"""One explainable B1 hit with a stable identity."""
"""One explainable strategy hit with a stable persistence identity."""
ts_code: str
name: str
target_trade_date: date
strategy: Literal["zhixing_b1"]
category: ZhixingB1Category
strategy: SelectionStrategyName
category: SelectionSignalCategory
close: float
details: Mapping[str, float | str | None] = field(
default_factory=lambda: dict[str, float | str | None]()
@@ -122,6 +133,7 @@ SelectionEvaluationStatus = Literal[
"no_signal",
"insufficient_history",
"missing_target_bar",
"missing_turnover_rate",
"data_error",
]
@@ -8,12 +8,17 @@ from datetime import date, datetime
from decimal import Decimal
from typing import Literal, Protocol
from .models import SelectionEvaluationStatus, SelectionSignal, StockHistory
from .models import (
SelectionEvaluationStatus,
SelectionSignal,
SelectionStrategyName,
StockHistory,
)
from .pattern_scoring import PatternScore
SelectionRunStatus = Literal["running", "success", "partial_success", "failed"]
SelectionRunItemStatus = SelectionEvaluationStatus
SelectionSignalCategoryFilter = Literal["pullback", "oversold", "original"]
SelectionSignalCategoryFilter = Literal["pullback", "oversold", "original", "resonance"]
SelectionResultSort = Literal["code", "score_desc", "score_asc"]
@@ -66,7 +71,7 @@ class SelectionRun:
"""A current execution attempt and its materialized result rows."""
id: str
strategy: Literal["zhixing_b1"]
strategy: SelectionStrategyName
target_trade_date: date
market_sync_batch_id: str | None
status: SelectionRunStatus
@@ -107,7 +112,7 @@ class SelectionRunStore(Protocol):
def prepare_run(
self,
strategy: Literal["zhixing_b1"],
strategy: SelectionStrategyName,
target_trade_date: date,
source: SelectionExecutionSource,
*,
@@ -138,7 +143,7 @@ class SelectionRunStore(Protocol):
def get_latest_run(
self,
strategy: Literal["zhixing_b1"],
strategy: SelectionStrategyName,
target_trade_date: date | None = None,
*,
query: SelectionResultQuery | None = None,
@@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from collections.abc import Generator, Sequence
from contextlib import contextmanager
from datetime import date, datetime
@@ -22,6 +23,9 @@ from ..domain.ports import MarketDataReaderError
from ..domain.runs import SelectionExecutionSource, SelectionStock
from .postgres_pool import SelectionConnectionPool, SelectionPostgresPool
# Use Uvicorn's configured logger so preflight diagnostics reach container logs.
logger = logging.getLogger("uvicorn.error.zhixing.selection.reader")
class SelectionReaderError(MarketDataReaderError):
"""Database read failure with stock and target-date context."""
@@ -40,10 +44,16 @@ SELECT
bar.high,
bar.low,
bar.close,
bar.vol
bar.vol,
basic.turnover_rate,
basic.total_mv
FROM market_daily_bar AS bar
LEFT JOIN market_stock AS stock
ON stock.ts_code = bar.ts_code
LEFT JOIN market_daily_basic AS basic
ON basic.ts_code = bar.ts_code
AND basic.trade_date = bar.trade_date
AND basic.trade_date = %s
WHERE bar.ts_code = ANY(%s)
AND bar.source_adj = 'qfq'
AND bar.trade_date <= %s
@@ -109,6 +119,25 @@ WHERE stock.is_active = true
ORDER BY stock.ts_code
"""
_GOLD_BRICK_ELIGIBLE_STOCKS_QUERY = """
SELECT stock.ts_code, stock.name
FROM market_stock AS stock
WHERE stock.is_active = true
AND EXISTS (
SELECT 1
FROM market_daily_bar AS bar
WHERE bar.ts_code = stock.ts_code
AND bar.trade_date = %s
AND bar.source_adj = 'qfq'
AND bar.open IS NOT NULL
AND bar.high IS NOT NULL
AND bar.low IS NOT NULL
AND bar.close IS NOT NULL
AND bar.vol IS NOT NULL
)
ORDER BY stock.ts_code
"""
_PATTERN_CASES_QUERY = """
WITH case_definition AS (
SELECT *
@@ -238,10 +267,9 @@ class PostgresMarketDataReader:
) -> tuple[StockHistory, ...]:
"""Read one bounded stock chunk with one parameterized qfq query.
Historical daily-basic values are deliberately not joined here: B1
only needs OHLCV for its historical formula. The execution-source
query still requires a complete target-day basic row before a stock is
admitted to a run.
Same-day daily-basic rows are left joined so strategies that require
target-day liquidity can fail closed without changing the OHLCV-only
Zhixing B1 formula.
"""
normalized = tuple(
@@ -255,7 +283,7 @@ class PostgresMarketDataReader:
with self._connection() as connection:
rows = connection.execute(
_HISTORY_QUERY,
(codes, target_trade_date),
(target_trade_date, codes, target_trade_date),
).fetchall()
except SelectionReaderError:
raise
@@ -273,9 +301,8 @@ class PostgresMarketDataReader:
"""Load the qualified market-data snapshot for a strategy run.
Args:
strategy: Supported strategy identity. The current reader accepts
``zhixing_b1`` and keeps the parameter explicit for future
strategy-specific eligibility rules.
strategy: Supported ``zhixing_b1`` or ``gold_brick`` identity,
used to apply strategy-specific target-day eligibility rules.
target_trade_date: Historical trading date to evaluate.
Returns:
@@ -287,19 +314,36 @@ class PostgresMarketDataReader:
SelectionReaderError: If PostgreSQL cannot complete the read.
"""
if strategy != "zhixing_b1":
if strategy not in {"zhixing_b1", "gold_brick"}:
raise SelectionMarketDataNotReady(f"unsupported selection strategy: {strategy}")
logger.info(
"selection_source_precheck_started strategy=%s target_trade_date=%s",
strategy,
target_trade_date.isoformat(),
)
try:
with self._connection() as connection:
source_row = connection.execute(_SOURCE_QUERY, (target_trade_date,)).fetchone()
if source_row is None:
logger.warning(
"selection_source_precheck_failed strategy=%s target_trade_date=%s "
"status=market_data_not_ready error_type=missing_eligible_sync_batch",
strategy,
target_trade_date.isoformat(),
)
raise SelectionMarketDataNotReady(
f"market data is not strategy-eligible for {target_trade_date.isoformat()}"
)
stock_rows = connection.execute(
_ELIGIBLE_STOCKS_QUERY,
(target_trade_date, target_trade_date),
).fetchall()
if strategy == "gold_brick":
stock_rows = connection.execute(
_GOLD_BRICK_ELIGIBLE_STOCKS_QUERY,
(target_trade_date,),
).fetchall()
else:
stock_rows = connection.execute(
_ELIGIBLE_STOCKS_QUERY,
(target_trade_date, target_trade_date),
).fetchall()
except (SelectionMarketDataNotReady, SelectionReaderError):
raise
except Exception as exc: # noqa: BLE001 - redact driver/pool details at the port boundary
@@ -311,9 +355,29 @@ class PostgresMarketDataReader:
SelectionStock(ts_code=str(row[0]), name=str(row[1] or "")) for row in stock_rows
)
if not stocks:
logger.warning(
"selection_source_precheck_failed strategy=%s target_trade_date=%s "
"market_sync_batch_id=%s status=market_data_not_ready "
"error_type=no_eligible_stocks",
strategy,
target_trade_date.isoformat(),
source_row[0],
)
raise SelectionMarketDataNotReady(
f"no eligible stocks have complete market data for {target_trade_date.isoformat()}"
)
logger.info(
"selection_source_precheck_ready strategy=%s target_trade_date=%s "
"market_sync_batch_id=%s target_count=%s valid_count=%s "
"eligible_count=%d coverage=%s status=ready",
strategy,
target_trade_date.isoformat(),
source_row[0],
source_row[1],
source_row[2],
len(stocks),
source_row[3],
)
return SelectionExecutionSource(
market_sync_batch_id=str(source_row[0]),
target_trade_date=target_trade_date,
@@ -14,7 +14,14 @@ from uuid import uuid4
import psycopg
from psycopg.types.json import Jsonb
from ..domain.models import SelectionSignal, ZhixingB1Category
from ..domain.gold_brick import GOLD_BRICK_SIGNAL_ORDER
from ..domain.models import (
GoldBrickCategory,
SelectionSignal,
SelectionSignalCategory,
SelectionStrategyName,
ZhixingB1Category,
)
from ..domain.pattern_scoring import (
ZHIXING_B1_PATTERN_CASES,
PatternScore,
@@ -35,19 +42,26 @@ from ..domain.runs import (
from ..domain.zhixing_b1 import ZHIXING_B1_SIGNAL_ORDER
from .postgres_pool import SelectionConnectionPool, SelectionPostgresPool
_SIGNAL_PRIORITY = {category: index for index, category in enumerate(ZHIXING_B1_SIGNAL_ORDER)}
_SELECTION_SIGNAL_ORDER: tuple[SelectionSignalCategory, ...] = (
*ZHIXING_B1_SIGNAL_ORDER,
*GOLD_BRICK_SIGNAL_ORDER,
)
_SIGNAL_PRIORITY = {
category: index for index, category in enumerate(_SELECTION_SIGNAL_ORDER)
}
_CATEGORY_PREFIXES = {
"pullback": "zhixing_b1_pullback_",
"oversold": "zhixing_b1_oversold_",
"original": "zhixing_b1_original_b1",
"resonance": "gold_brick_resonance",
}
_SIGNAL_ORDER_SQL = (
"CASE category "
+ " ".join(
f"WHEN '{category.value}' THEN {index}"
for index, category in enumerate(ZHIXING_B1_SIGNAL_ORDER)
for index, category in enumerate(_SELECTION_SIGNAL_ORDER)
)
+ f" ELSE {len(ZHIXING_B1_SIGNAL_ORDER)} END"
+ f" ELSE {len(_SELECTION_SIGNAL_ORDER)} END"
)
_PATTERN_CASES_BY_ID = {definition.id: definition for definition in ZHIXING_B1_PATTERN_CASES}
_STOCK_ORDER_SQL = {
@@ -120,7 +134,7 @@ class PostgresSelectionRunRepository(SelectionRunStore):
def prepare_run(
self,
strategy: Literal["zhixing_b1"],
strategy: SelectionStrategyName,
target_trade_date: date,
source: SelectionExecutionSource,
*,
@@ -332,7 +346,7 @@ class PostgresSelectionRunRepository(SelectionRunStore):
def get_latest_run(
self,
strategy: Literal["zhixing_b1"],
strategy: SelectionStrategyName,
target_trade_date: date | None = None,
*,
query: SelectionResultQuery | None = None,
@@ -463,6 +477,7 @@ class PostgresSelectionRunRepository(SelectionRunStore):
"no_signal",
"insufficient_history",
"missing_target_bar",
"missing_turnover_rate",
"data_error",
],
str(value[2]),
@@ -476,7 +491,7 @@ class PostgresSelectionRunRepository(SelectionRunStore):
)
return SelectionRun(
id=str(row[0]),
strategy=cast(Literal["zhixing_b1"], str(row[1])),
strategy=cast(SelectionStrategyName, str(row[1])),
target_trade_date=_as_date(row[2]),
market_sync_batch_id=str(row[3]) if row[3] is not None else None,
status=cast(SelectionRunStatus, str(row[4])),
@@ -537,13 +552,22 @@ def _signal_from_row(row: tuple[object, ...]) -> SelectionSignal:
ts_code=str(row[0]),
name=str(row[1] or ""),
target_trade_date=_as_date(row[2]),
strategy=cast(Literal["zhixing_b1"], str(row[3])),
category=ZhixingB1Category(str(row[4])),
strategy=cast(SelectionStrategyName, str(row[3])),
category=_signal_category(str(row[4])),
close=float(str(row[5])),
details=_details(row[6]),
)
def _signal_category(value: str) -> SelectionSignalCategory:
"""Map a persisted category for either supported selection strategy."""
try:
return ZhixingB1Category(value)
except ValueError:
return GoldBrickCategory(value)
def _stock_filter(query: SelectionResultQuery, run_id: str) -> tuple[str, list[object]]:
"""Build the signal predicate used to select distinct matching stocks.
@@ -14,10 +14,16 @@ from zhixing_server.modules.selection.application.chart import (
SelectionChart,
SelectionChartNotFound,
)
from zhixing_server.modules.selection.application.evaluate_gold_brick import (
EvaluateGoldBrick,
)
from zhixing_server.modules.selection.application.run import (
RunZhixingB1,
)
from zhixing_server.modules.selection.domain.models import SelectionSignal
from zhixing_server.modules.selection.domain.models import (
SelectionSignal,
SelectionStrategyName,
)
from zhixing_server.modules.selection.domain.pattern_scoring import (
PatternScore,
ZhixingB1PatternScorer,
@@ -45,7 +51,7 @@ selection_router = APIRouter()
_SELECTION_POOL_CACHE_LOCK = threading.Lock()
_SELECTION_POOL_CACHE: dict[tuple[str, int], SelectionPostgresPool] = {}
StrategyValue = Literal["zhixing_b1"]
StrategyValue = SelectionStrategyName
SelectionStatusValue = Literal[
"no_data",
"running",
@@ -216,6 +222,7 @@ def get_selection_service(
return RunZhixingB1(
reader,
store,
evaluators={"gold_brick": EvaluateGoldBrick(reader)},
pattern_case_loader=pattern_case_loader,
pattern_scorer=ZhixingB1PatternScorer(),
pattern_scoring_enabled=settings.selection_pattern_scoring_enabled,
@@ -324,7 +331,9 @@ def get_selection_run(
page: Annotated[int, Query(ge=1)] = 1,
page_size: Annotated[int, Query(ge=1, le=100)] = 10,
search: Annotated[str | None, Query(max_length=100)] = None,
category: Literal["pullback", "oversold", "original"] | None = None,
category: Literal[
"pullback", "oversold", "original", "resonance"
] | None = None,
sort: Literal["code", "score_desc", "score_asc"] = "code",
) -> SelectionResultsResponse:
"""Return one run for asynchronous polling."""
@@ -347,7 +356,9 @@ def get_selection_results(
page: Annotated[int, Query(ge=1)] = 1,
page_size: Annotated[int, Query(ge=1, le=100)] = 10,
search: Annotated[str | None, Query(max_length=100)] = None,
category: Literal["pullback", "oversold", "original"] | None = None,
category: Literal[
"pullback", "oversold", "original", "resonance"
] | None = None,
sort: Literal["code", "score_desc", "score_asc"] = "code",
) -> SelectionResultsResponse:
"""Return the current persisted result for a strategy and optional date."""
@@ -410,7 +421,13 @@ def _run_response(run: SelectionRun, *, query: SelectionResultQuery) -> Selectio
reason=item.reason,
)
for item in run.items
if item.status in {"insufficient_history", "missing_target_bar", "data_error"}
if item.status
in {
"insufficient_history",
"missing_target_bar",
"missing_turnover_rate",
"data_error",
}
],
stocks=[
SelectionStockResponse(
@@ -496,7 +513,7 @@ def _result_query(
page: int,
page_size: int,
search: str | None,
category: Literal["pullback", "oversold", "original"] | None,
category: Literal["pullback", "oversold", "original", "resonance"] | None,
sort: Literal["code", "score_desc", "score_asc"],
) -> SelectionResultQuery:
"""Normalize HTTP query values before handing them to the selection port."""