130 lines
4.1 KiB
Python
130 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from backend.bootstrap.config import display_compact_date as _display_date
|
|
from backend.data.numbers import finite_number as _number
|
|
from backend.features.screener.backtest import BacktestRunner
|
|
from backend.features.screener.catalog import (
|
|
ADVANCED_CURATED_STRATEGIES,
|
|
ALLOWED_OPERATORS,
|
|
BUILTIN_STRATEGIES,
|
|
CURATED_STRATEGIES,
|
|
FACTOR_FIELDS,
|
|
FACTOR_GROUPS,
|
|
REGIMES,
|
|
STRATEGY_ENVIRONMENT_NOTES,
|
|
)
|
|
from backend.features.screener.data_sync import (
|
|
FactorDataService,
|
|
_earnings_event_rows,
|
|
_popularity_factor_rows,
|
|
_quarter_periods,
|
|
)
|
|
from backend.features.screener.factors import FactorBuilder
|
|
from backend.features.screener.formula import FormulaEvaluator, compile_local_strategy
|
|
from backend.features.screener.indicators import (
|
|
_available_percentile_map,
|
|
_broken_reversal_metrics,
|
|
_ema,
|
|
_ending_streak,
|
|
_is_limit_bar,
|
|
_limit_threshold,
|
|
_macd_last,
|
|
_macd_series,
|
|
_matches,
|
|
_max_streak,
|
|
_optional_number,
|
|
_pearson,
|
|
_percentile_map,
|
|
_regime_reason,
|
|
_risk_flags,
|
|
_rounded_optional,
|
|
_rsi,
|
|
_touched_limit_bar,
|
|
_weekly_series,
|
|
)
|
|
from backend.features.screener.regime import RegimeDetector
|
|
from backend.features.screener.selection import SelectionRunner
|
|
from database import ReviewDatabase
|
|
|
|
|
|
class ScreenerEngine:
|
|
"""Stable facade over the independently owned screener services."""
|
|
|
|
def __init__(self, database: ReviewDatabase) -> None:
|
|
self.database = database
|
|
self.factor_builder = FactorBuilder(database)
|
|
self.formula_evaluator = FormulaEvaluator()
|
|
self.regime_detector = RegimeDetector(database)
|
|
self.backtest_runner = BacktestRunner(
|
|
database, self.factor_builder, self.formula_evaluator
|
|
)
|
|
self.selection_runner = SelectionRunner(
|
|
database,
|
|
self.factor_builder,
|
|
self.formula_evaluator,
|
|
self.backtest_runner,
|
|
)
|
|
|
|
def ensure_builtin_strategies(self) -> None:
|
|
existing = {
|
|
item["name"]: item
|
|
for item in self.database.list_screener_strategies()
|
|
if item["builtin"]
|
|
}
|
|
for strategy in BUILTIN_STRATEGIES:
|
|
current = existing.get(strategy["name"])
|
|
self.database.save_screener_strategy(
|
|
None, **strategy, builtin=True,
|
|
strategy_id=int(current["id"]) if current else None,
|
|
)
|
|
|
|
def detect_regime(self, trade_date: str) -> dict[str, Any]:
|
|
return self.regime_detector.detect_regime(trade_date)
|
|
|
|
def factor_health(self, trade_date: str) -> dict[str, Any]:
|
|
return self.database.factor_health_summary(trade_date)
|
|
|
|
def validate_formula(self, formula: dict[str, Any]) -> dict[str, Any]:
|
|
return self.formula_evaluator.validate_formula(formula)
|
|
|
|
def screen(
|
|
self, user_id: int, trade_date: str, formula: dict[str, Any], regime: str,
|
|
strategy_name: str, run_backtest: bool = True,
|
|
realtime_snapshot: dict[str, Any] | None = None,
|
|
mode: str = "smart",
|
|
prepared_factors: list[dict[str, Any]] | None = None,
|
|
prepared_date: str = "",
|
|
) -> dict[str, Any]:
|
|
return self.selection_runner.screen(
|
|
user_id,
|
|
trade_date,
|
|
formula,
|
|
regime,
|
|
strategy_name,
|
|
run_backtest,
|
|
realtime_snapshot,
|
|
mode,
|
|
prepared_factors,
|
|
prepared_date,
|
|
)
|
|
|
|
def build_factors(
|
|
self,
|
|
trade_date: str,
|
|
realtime_snapshot: dict[str, Any] | None = None,
|
|
history_days: int = 80,
|
|
) -> tuple[list[dict[str, Any]], str]:
|
|
return self.factor_builder.build_factors(
|
|
trade_date, realtime_snapshot, history_days
|
|
)
|
|
|
|
def apply_formula(
|
|
self, rows: list[dict[str, Any]], formula: dict[str, Any], regime: str
|
|
) -> list[dict[str, Any]]:
|
|
return self.formula_evaluator.apply_formula(rows, formula, regime)
|
|
|
|
def backtest(self, trade_date: str, formula: dict[str, Any]) -> dict[str, Any]:
|
|
return self.backtest_runner.backtest(trade_date, formula)
|