87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
|
"""
|
||
|
|
Strategy optimizer schemas.
|
||
|
|
"""
|
||
|
|
from datetime import date
|
||
|
|
from decimal import Decimal
|
||
|
|
from typing import Annotated, Any, Dict, List, Optional
|
||
|
|
|
||
|
|
from pydantic import BaseModel, Field, PlainSerializer
|
||
|
|
|
||
|
|
FloatDecimal = Annotated[
|
||
|
|
Decimal,
|
||
|
|
PlainSerializer(lambda v: float(v), return_type=float, when_used="json"),
|
||
|
|
]
|
||
|
|
|
||
|
|
# --- Default parameter grids per strategy type ---
|
||
|
|
|
||
|
|
KJB_DEFAULT_GRID: Dict[str, List[Any]] = {
|
||
|
|
"stop_loss_pct": [0.03, 0.05, 0.07],
|
||
|
|
"target1_pct": [0.05, 0.07, 0.10],
|
||
|
|
"rs_lookback": [10, 20, 30],
|
||
|
|
}
|
||
|
|
|
||
|
|
MULTI_FACTOR_DEFAULT_GRID: Dict[str, List[Any]] = {
|
||
|
|
"weights.value": [0.15, 0.25, 0.35],
|
||
|
|
"weights.quality": [0.15, 0.25, 0.35],
|
||
|
|
"weights.momentum": [0.15, 0.25, 0.35],
|
||
|
|
}
|
||
|
|
|
||
|
|
QUALITY_DEFAULT_GRID: Dict[str, List[Any]] = {
|
||
|
|
"min_fscore": [5, 6, 7, 8],
|
||
|
|
}
|
||
|
|
|
||
|
|
VALUE_MOMENTUM_DEFAULT_GRID: Dict[str, List[Any]] = {
|
||
|
|
"value_weight": [0.3, 0.4, 0.5, 0.6, 0.7],
|
||
|
|
"momentum_weight": [0.3, 0.4, 0.5, 0.6, 0.7],
|
||
|
|
}
|
||
|
|
|
||
|
|
DEFAULT_GRIDS: Dict[str, Dict[str, List[Any]]] = {
|
||
|
|
"kjb": KJB_DEFAULT_GRID,
|
||
|
|
"multi_factor": MULTI_FACTOR_DEFAULT_GRID,
|
||
|
|
"quality": QUALITY_DEFAULT_GRID,
|
||
|
|
"value_momentum": VALUE_MOMENTUM_DEFAULT_GRID,
|
||
|
|
}
|
||
|
|
|
||
|
|
STRATEGY_TYPES = ["kjb", "multi_factor", "quality", "value_momentum"]
|
||
|
|
|
||
|
|
|
||
|
|
class OptimizeRequest(BaseModel):
|
||
|
|
strategy_type: str = Field(
|
||
|
|
...,
|
||
|
|
description="Strategy type: kjb, multi_factor, quality, value_momentum",
|
||
|
|
)
|
||
|
|
param_grid: Optional[Dict[str, List[Any]]] = Field(
|
||
|
|
default=None,
|
||
|
|
description="Parameter grid. If None, uses default preset for the strategy type.",
|
||
|
|
)
|
||
|
|
start_date: date
|
||
|
|
end_date: date
|
||
|
|
initial_capital: Decimal = Field(default=Decimal("100000000"), gt=0)
|
||
|
|
commission_rate: Decimal = Field(default=Decimal("0.00015"), ge=0, le=1)
|
||
|
|
slippage_rate: Decimal = Field(default=Decimal("0.001"), ge=0, le=1)
|
||
|
|
benchmark: str = Field(default="KOSPI")
|
||
|
|
top_n: int = Field(default=30, ge=1, le=100)
|
||
|
|
rank_by: str = Field(
|
||
|
|
default="sharpe_ratio",
|
||
|
|
description="Metric to rank results by: sharpe_ratio, cagr, total_return, mdd",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class OptimizeResultItem(BaseModel):
|
||
|
|
rank: int
|
||
|
|
params: Dict[str, Any]
|
||
|
|
total_return: FloatDecimal
|
||
|
|
cagr: FloatDecimal
|
||
|
|
mdd: FloatDecimal
|
||
|
|
sharpe_ratio: FloatDecimal
|
||
|
|
volatility: FloatDecimal
|
||
|
|
benchmark_return: FloatDecimal
|
||
|
|
excess_return: FloatDecimal
|
||
|
|
|
||
|
|
|
||
|
|
class OptimizeResponse(BaseModel):
|
||
|
|
strategy_type: str
|
||
|
|
total_combinations: int
|
||
|
|
results: List[OptimizeResultItem]
|
||
|
|
best_params: Dict[str, Any]
|