79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
"""
|
|
Quant strategy API endpoints.
|
|
"""
|
|
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.schemas.strategy import (
|
|
MultiFactorRequest, QualityRequest, ValueMomentumRequest, KJBRequest, StrategyResult,
|
|
)
|
|
from app.services.strategy import MultiFactorStrategy, QualityStrategy, ValueMomentumStrategy, KJBStrategy
|
|
|
|
router = APIRouter(prefix="/api/strategy", tags=["strategy"])
|
|
|
|
|
|
@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)
|
|
return strategy.run(
|
|
universe_filter=request.universe,
|
|
top_n=request.top_n,
|
|
base_date=request.base_date,
|
|
weights=request.weights,
|
|
)
|
|
|
|
|
|
@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)
|
|
return strategy.run(
|
|
universe_filter=request.universe,
|
|
top_n=request.top_n,
|
|
base_date=request.base_date,
|
|
min_fscore=request.min_fscore,
|
|
)
|
|
|
|
|
|
@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)
|
|
return 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,
|
|
)
|
|
|
|
|
|
@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)
|
|
return strategy.run(
|
|
universe_filter=request.universe,
|
|
top_n=request.top_n,
|
|
base_date=request.base_date,
|
|
)
|