4483f6e4ba
Add dc_only parameter to all strategy endpoints. When true, filters results to include only tickers present in the ETF table, supporting DC pension investment constraints where only ETFs are allowed. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
"""
|
|
Quant strategy API endpoints.
|
|
"""
|
|
from typing import Set
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.database import get_db
|
|
from app.api.deps import CurrentUser
|
|
from app.models.stock import ETF
|
|
from app.schemas.strategy import (
|
|
MultiFactorRequest, QualityRequest, ValueMomentumRequest, KJBRequest, StrategyResult,
|
|
)
|
|
from app.services.strategy import MultiFactorStrategy, QualityStrategy, ValueMomentumStrategy, KJBStrategy
|
|
|
|
router = APIRouter(prefix="/api/strategy", tags=["strategy"])
|
|
|
|
|
|
def _filter_dc_only(result: StrategyResult, db: Session) -> StrategyResult:
|
|
"""Filter strategy result to include only ETFs (DC pension investable)."""
|
|
tickers = [s.ticker for s in result.stocks]
|
|
etf_tickers: Set[str] = set(
|
|
row[0] for row in db.query(ETF.ticker).filter(ETF.ticker.in_(tickers)).all()
|
|
) if tickers else set()
|
|
|
|
filtered = [s for s in result.stocks if s.ticker in etf_tickers]
|
|
# Re-rank
|
|
for i, stock in enumerate(filtered, 1):
|
|
stock.rank = i
|
|
|
|
return StrategyResult(
|
|
strategy_name=result.strategy_name,
|
|
base_date=result.base_date,
|
|
universe_count=result.universe_count,
|
|
result_count=len(filtered),
|
|
stocks=filtered,
|
|
)
|
|
|
|
|
|
@router.post("/multi-factor", response_model=StrategyResult)
|
|
async def run_multi_factor(
|
|
request: MultiFactorRequest,
|
|
current_user: CurrentUser,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Run multi-factor strategy."""
|
|
strategy = MultiFactorStrategy(db)
|
|
result = strategy.run(
|
|
universe_filter=request.universe,
|
|
top_n=request.top_n,
|
|
base_date=request.base_date,
|
|
weights=request.weights,
|
|
)
|
|
return _filter_dc_only(result, db) if request.dc_only else result
|
|
|
|
|
|
@router.post("/quality", response_model=StrategyResult)
|
|
async def run_quality(
|
|
request: QualityRequest,
|
|
current_user: CurrentUser,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Run super quality strategy."""
|
|
strategy = QualityStrategy(db)
|
|
result = strategy.run(
|
|
universe_filter=request.universe,
|
|
top_n=request.top_n,
|
|
base_date=request.base_date,
|
|
min_fscore=request.min_fscore,
|
|
)
|
|
return _filter_dc_only(result, db) if request.dc_only else result
|
|
|
|
|
|
@router.post("/value-momentum", response_model=StrategyResult)
|
|
async def run_value_momentum(
|
|
request: ValueMomentumRequest,
|
|
current_user: CurrentUser,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Run value-momentum strategy."""
|
|
strategy = ValueMomentumStrategy(db)
|
|
result = strategy.run(
|
|
universe_filter=request.universe,
|
|
top_n=request.top_n,
|
|
base_date=request.base_date,
|
|
value_weight=request.value_weight,
|
|
momentum_weight=request.momentum_weight,
|
|
)
|
|
return _filter_dc_only(result, db) if request.dc_only else result
|
|
|
|
|
|
@router.post("/kjb", response_model=StrategyResult)
|
|
async def run_kjb(
|
|
request: KJBRequest,
|
|
current_user: CurrentUser,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Run KJB strategy."""
|
|
strategy = KJBStrategy(db)
|
|
result = strategy.run(
|
|
universe_filter=request.universe,
|
|
top_n=request.top_n,
|
|
base_date=request.base_date,
|
|
)
|
|
return _filter_dc_only(result, db) if request.dc_only else result
|