zephyrdark
3f8ef7e108
- BaseStrategy abstract class - MultiFactorStrategy with weighted factors - QualityStrategy with F-Score filtering - ValueMomentumStrategy combining value and momentum Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
"""
|
|
Multi-factor strategy implementation.
|
|
"""
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from typing import List
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.services.strategy.base import BaseStrategy
|
|
from app.schemas.strategy import (
|
|
StockFactor, StrategyResult, UniverseFilter, FactorWeights,
|
|
)
|
|
|
|
|
|
class MultiFactorStrategy(BaseStrategy):
|
|
"""Multi-factor strategy combining value, quality, and momentum."""
|
|
|
|
strategy_name = "multi_factor"
|
|
|
|
def run(
|
|
self,
|
|
universe_filter: UniverseFilter,
|
|
top_n: int,
|
|
base_date: date = None,
|
|
weights: FactorWeights = None,
|
|
) -> StrategyResult:
|
|
if base_date is None:
|
|
base_date = date.today()
|
|
if weights is None:
|
|
weights = FactorWeights()
|
|
|
|
# Get universe
|
|
stocks = self.get_universe(universe_filter)
|
|
tickers = [s.ticker for s in stocks]
|
|
stock_map = {s.ticker: s for s in stocks}
|
|
|
|
if not tickers:
|
|
return StrategyResult(
|
|
strategy_name=self.strategy_name,
|
|
base_date=base_date,
|
|
universe_count=0,
|
|
result_count=0,
|
|
stocks=[],
|
|
)
|
|
|
|
# Get valuations and sectors
|
|
valuations = self.factor_calc.get_valuations(tickers, base_date)
|
|
sectors = self.factor_calc.get_sectors(tickers)
|
|
|
|
# Calculate factor scores
|
|
value_scores = self.factor_calc.calculate_value_scores(valuations)
|
|
quality_scores = self.factor_calc.calculate_quality_scores(tickers, base_date)
|
|
momentum = self.factor_calc.calculate_momentum(tickers, base_date)
|
|
|
|
# Normalize momentum to z-scores
|
|
if momentum:
|
|
mom_values = list(momentum.values())
|
|
mom_mean = sum(mom_values) / len(mom_values)
|
|
mom_std = (sum((v - mom_mean) ** 2 for v in mom_values) / len(mom_values)) ** 0.5
|
|
if mom_std > 0:
|
|
momentum_scores = {
|
|
t: Decimal(str((float(v) - float(mom_mean)) / float(mom_std)))
|
|
for t, v in momentum.items()
|
|
}
|
|
else:
|
|
momentum_scores = {t: Decimal("0") for t in momentum}
|
|
else:
|
|
momentum_scores = {}
|
|
|
|
# Build result
|
|
results = []
|
|
for ticker in tickers:
|
|
stock = stock_map[ticker]
|
|
val = valuations.get(ticker)
|
|
|
|
v_score = value_scores.get(ticker, Decimal("0"))
|
|
q_score = quality_scores.get(ticker, Decimal("0"))
|
|
m_score = momentum_scores.get(ticker, Decimal("0"))
|
|
|
|
# Weighted composite
|
|
total = (
|
|
v_score * weights.value +
|
|
q_score * weights.quality +
|
|
m_score * weights.momentum
|
|
)
|
|
|
|
results.append(StockFactor(
|
|
ticker=ticker,
|
|
name=stock.name,
|
|
market=stock.market,
|
|
sector_name=sectors.get(ticker),
|
|
market_cap=int(stock.market_cap / 100_000_000) if stock.market_cap else None,
|
|
close_price=Decimal(str(stock.close_price)) if stock.close_price else None,
|
|
per=Decimal(str(val.per)) if val and val.per else None,
|
|
pbr=Decimal(str(val.pbr)) if val and val.pbr else None,
|
|
psr=Decimal(str(val.psr)) if val and val.psr else None,
|
|
pcr=Decimal(str(val.pcr)) if val and val.pcr else None,
|
|
dividend_yield=Decimal(str(val.dividend_yield)) if val and val.dividend_yield else None,
|
|
value_score=v_score,
|
|
quality_score=q_score,
|
|
momentum_score=m_score,
|
|
total_score=total,
|
|
))
|
|
|
|
# Sort by total score descending
|
|
results.sort(key=lambda x: x.total_score or Decimal("0"), reverse=True)
|
|
|
|
# Assign ranks and limit
|
|
for i, r in enumerate(results[:top_n], 1):
|
|
r.rank = i
|
|
|
|
return StrategyResult(
|
|
strategy_name=self.strategy_name,
|
|
base_date=base_date,
|
|
universe_count=len(stocks),
|
|
result_count=min(top_n, len(results)),
|
|
stocks=results[:top_n],
|
|
)
|