From 3f8ef7e10802b60551726389087a88cd6924b968 Mon Sep 17 00:00:00 2001 From: zephyrdark Date: Tue, 3 Feb 2026 08:59:05 +0900 Subject: [PATCH] feat: add multi-factor, quality, and value-momentum strategies - BaseStrategy abstract class - MultiFactorStrategy with weighted factors - QualityStrategy with F-Score filtering - ValueMomentumStrategy combining value and momentum Co-Authored-By: Claude Opus 4.5 --- backend/app/services/strategy/__init__.py | 6 + backend/app/services/strategy/base.py | 42 +++++++ backend/app/services/strategy/multi_factor.py | 119 ++++++++++++++++++ backend/app/services/strategy/quality.py | 84 +++++++++++++ .../app/services/strategy/value_momentum.py | 98 +++++++++++++++ 5 files changed, 349 insertions(+) create mode 100644 backend/app/services/strategy/__init__.py create mode 100644 backend/app/services/strategy/base.py create mode 100644 backend/app/services/strategy/multi_factor.py create mode 100644 backend/app/services/strategy/quality.py create mode 100644 backend/app/services/strategy/value_momentum.py diff --git a/backend/app/services/strategy/__init__.py b/backend/app/services/strategy/__init__.py new file mode 100644 index 0000000..8c5de1c --- /dev/null +++ b/backend/app/services/strategy/__init__.py @@ -0,0 +1,6 @@ +from app.services.strategy.base import BaseStrategy +from app.services.strategy.multi_factor import MultiFactorStrategy +from app.services.strategy.quality import QualityStrategy +from app.services.strategy.value_momentum import ValueMomentumStrategy + +__all__ = ["BaseStrategy", "MultiFactorStrategy", "QualityStrategy", "ValueMomentumStrategy"] diff --git a/backend/app/services/strategy/base.py b/backend/app/services/strategy/base.py new file mode 100644 index 0000000..71a4030 --- /dev/null +++ b/backend/app/services/strategy/base.py @@ -0,0 +1,42 @@ +""" +Base strategy class. +""" +from abc import ABC, abstractmethod +from datetime import date +from typing import List + +from sqlalchemy.orm import Session + +from app.schemas.strategy import StockFactor, StrategyResult, UniverseFilter +from app.services.factor_calculator import FactorCalculator + + +class BaseStrategy(ABC): + """Base class for quant strategies.""" + + strategy_name: str = "base" + + def __init__(self, db: Session): + self.db = db + self.factor_calc = FactorCalculator(db) + + def get_universe(self, filter: UniverseFilter) -> List: + """Get filtered stock universe.""" + return self.factor_calc.get_universe( + markets=filter.markets, + min_market_cap=filter.min_market_cap, + max_market_cap=filter.max_market_cap, + exclude_stock_types=filter.exclude_stock_types, + exclude_sectors=filter.exclude_sectors, + ) + + @abstractmethod + def run( + self, + universe_filter: UniverseFilter, + top_n: int, + base_date: date = None, + **kwargs, + ) -> StrategyResult: + """Run the strategy and return ranked stocks.""" + pass diff --git a/backend/app/services/strategy/multi_factor.py b/backend/app/services/strategy/multi_factor.py new file mode 100644 index 0000000..5a2c7a0 --- /dev/null +++ b/backend/app/services/strategy/multi_factor.py @@ -0,0 +1,119 @@ +""" +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], + ) diff --git a/backend/app/services/strategy/quality.py b/backend/app/services/strategy/quality.py new file mode 100644 index 0000000..ed54102 --- /dev/null +++ b/backend/app/services/strategy/quality.py @@ -0,0 +1,84 @@ +""" +Super Quality strategy implementation. +""" +from datetime import date +from decimal import Decimal + +from app.services.strategy.base import BaseStrategy +from app.schemas.strategy import StockFactor, StrategyResult, UniverseFilter + + +class QualityStrategy(BaseStrategy): + """Super Quality strategy using F-Score and quality factors.""" + + strategy_name = "quality" + + def run( + self, + universe_filter: UniverseFilter, + top_n: int, + base_date: date = None, + min_fscore: int = 7, + ) -> StrategyResult: + if base_date is None: + base_date = date.today() + + # 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 data + valuations = self.factor_calc.get_valuations(tickers, base_date) + sectors = self.factor_calc.get_sectors(tickers) + fscores = self.factor_calc.calculate_fscore(tickers) + quality_scores = self.factor_calc.calculate_quality_scores(tickers, base_date) + + # Filter by F-Score + qualified_tickers = [t for t in tickers if fscores.get(t, 0) >= min_fscore] + + # Build results + results = [] + for ticker in qualified_tickers: + stock = stock_map[ticker] + val = valuations.get(ticker) + q_score = quality_scores.get(ticker, Decimal("0")) + fscore = fscores.get(ticker, 0) + + 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, + dividend_yield=Decimal(str(val.dividend_yield)) if val and val.dividend_yield else None, + quality_score=q_score, + total_score=q_score, + fscore=fscore, + )) + + # Sort by quality score + results.sort(key=lambda x: x.total_score or Decimal("0"), reverse=True) + + 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], + ) diff --git a/backend/app/services/strategy/value_momentum.py b/backend/app/services/strategy/value_momentum.py new file mode 100644 index 0000000..bb6ca7e --- /dev/null +++ b/backend/app/services/strategy/value_momentum.py @@ -0,0 +1,98 @@ +""" +Value-Momentum strategy implementation. +""" +from datetime import date +from decimal import Decimal + +from app.services.strategy.base import BaseStrategy +from app.schemas.strategy import StockFactor, StrategyResult, UniverseFilter + + +class ValueMomentumStrategy(BaseStrategy): + """Value-Momentum combined strategy.""" + + strategy_name = "value_momentum" + + def run( + self, + universe_filter: UniverseFilter, + top_n: int, + base_date: date = None, + value_weight: Decimal = Decimal("0.5"), + momentum_weight: Decimal = Decimal("0.5"), + ) -> StrategyResult: + if base_date is None: + base_date = date.today() + + # 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 data + valuations = self.factor_calc.get_valuations(tickers, base_date) + sectors = self.factor_calc.get_sectors(tickers) + value_scores = self.factor_calc.calculate_value_scores(valuations) + momentum = self.factor_calc.calculate_momentum(tickers, base_date) + + # Normalize momentum + 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 results + results = [] + for ticker in tickers: + stock = stock_map[ticker] + val = valuations.get(ticker) + v_score = value_scores.get(ticker, Decimal("0")) + m_score = momentum_scores.get(ticker, Decimal("0")) + + total = v_score * value_weight + m_score * momentum_weight + + 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, + dividend_yield=Decimal(str(val.dividend_yield)) if val and val.dividend_yield else None, + value_score=v_score, + momentum_score=m_score, + total_score=total, + )) + + results.sort(key=lambda x: x.total_score or Decimal("0"), reverse=True) + + 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], + )