quant/galaxis-po
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add multi-factor, quality, and value-momentum strategies

Browse Source 
- 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>
This commit is contained in:
zephyrdark 2026-02-03 08:59:05 +09:00
parent 9eebc73390
commit 3f8ef7e108
5 changed files with 349 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
backend/app/services/strategy/__init__.py Normal file
View File

@ -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"]

42
backend/app/services/strategy/base.py Normal file
View File

@ -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

119
backend/app/services/strategy/multi_factor.py Normal file
View File

@ -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],
)

84
backend/app/services/strategy/quality.py Normal file
View File

@ -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],
)

98
backend/app/services/strategy/value_momentum.py Normal file
View File

@ -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],
)