162 lines
4.7 KiB
Python
162 lines
4.7 KiB
Python
"""Backtest service."""
|
|
from typing import Dict, Any
|
|
from datetime import datetime
|
|
from sqlalchemy.orm import Session
|
|
from uuid import UUID
|
|
|
|
from app.models.backtest import BacktestRun, BacktestTrade
|
|
from app.backtest.engine import BacktestEngine
|
|
from app.strategies import get_strategy
|
|
from app.schemas.backtest import BacktestConfig
|
|
|
|
|
|
class BacktestService:
|
|
"""백테스트 서비스."""
|
|
|
|
@staticmethod
|
|
def run_backtest(config: BacktestConfig, db_session: Session) -> BacktestRun:
|
|
"""
|
|
백테스트 실행.
|
|
|
|
Args:
|
|
config: 백테스트 설정
|
|
db_session: 데이터베이스 세션
|
|
|
|
Returns:
|
|
백테스트 실행 레코드
|
|
"""
|
|
# 백테스트 실행 레코드 생성
|
|
backtest_run = BacktestRun(
|
|
name=config.name,
|
|
strategy_name=config.strategy_name,
|
|
start_date=config.start_date,
|
|
end_date=config.end_date,
|
|
initial_capital=config.initial_capital,
|
|
status='running',
|
|
config=config.strategy_config or {}
|
|
)
|
|
db_session.add(backtest_run)
|
|
db_session.commit()
|
|
db_session.refresh(backtest_run)
|
|
|
|
try:
|
|
# 전략 인스턴스 생성
|
|
strategy = get_strategy(
|
|
strategy_name=config.strategy_name,
|
|
config=config.strategy_config
|
|
)
|
|
|
|
# 백테스트 엔진 생성
|
|
engine = BacktestEngine(
|
|
initial_capital=config.initial_capital,
|
|
commission_rate=config.commission_rate,
|
|
rebalance_frequency=config.rebalance_frequency
|
|
)
|
|
|
|
# 백테스트 실행
|
|
results = engine.run(
|
|
strategy=strategy,
|
|
start_date=datetime.combine(config.start_date, datetime.min.time()),
|
|
end_date=datetime.combine(config.end_date, datetime.min.time()),
|
|
db_session=db_session
|
|
)
|
|
|
|
# 결과 저장
|
|
backtest_run.status = 'completed'
|
|
backtest_run.results = results
|
|
|
|
# 거래 내역 저장
|
|
for trade_data in results['trades']:
|
|
trade = BacktestTrade(
|
|
backtest_run_id=backtest_run.id,
|
|
ticker=trade_data['ticker'],
|
|
trade_date=trade_data['date'],
|
|
action=trade_data['action'],
|
|
quantity=trade_data['quantity'],
|
|
price=trade_data['price'],
|
|
commission=0, # TODO: 수수료 계산
|
|
pnl=trade_data.get('pnl')
|
|
)
|
|
db_session.add(trade)
|
|
|
|
db_session.commit()
|
|
db_session.refresh(backtest_run)
|
|
|
|
except Exception as e:
|
|
print(f"백테스트 실행 오류: {e}")
|
|
backtest_run.status = 'failed'
|
|
backtest_run.results = {'error': str(e)}
|
|
db_session.commit()
|
|
db_session.refresh(backtest_run)
|
|
|
|
return backtest_run
|
|
|
|
@staticmethod
|
|
def get_backtest(backtest_id: UUID, db_session: Session) -> BacktestRun:
|
|
"""
|
|
백테스트 조회.
|
|
|
|
Args:
|
|
backtest_id: 백테스트 ID
|
|
db_session: 데이터베이스 세션
|
|
|
|
Returns:
|
|
백테스트 실행 레코드
|
|
"""
|
|
backtest_run = db_session.query(BacktestRun).filter(
|
|
BacktestRun.id == backtest_id
|
|
).first()
|
|
|
|
return backtest_run
|
|
|
|
@staticmethod
|
|
def list_backtests(
|
|
db_session: Session,
|
|
skip: int = 0,
|
|
limit: int = 100
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
백테스트 목록 조회.
|
|
|
|
Args:
|
|
db_session: 데이터베이스 세션
|
|
skip: 건너뛸 레코드 수
|
|
limit: 최대 레코드 수
|
|
|
|
Returns:
|
|
백테스트 목록
|
|
"""
|
|
total = db_session.query(BacktestRun).count()
|
|
items = db_session.query(BacktestRun).order_by(
|
|
BacktestRun.created_at.desc()
|
|
).offset(skip).limit(limit).all()
|
|
|
|
return {
|
|
'items': items,
|
|
'total': total
|
|
}
|
|
|
|
@staticmethod
|
|
def delete_backtest(backtest_id: UUID, db_session: Session) -> bool:
|
|
"""
|
|
백테스트 삭제.
|
|
|
|
Args:
|
|
backtest_id: 백테스트 ID
|
|
db_session: 데이터베이스 세션
|
|
|
|
Returns:
|
|
삭제 성공 여부
|
|
"""
|
|
backtest_run = db_session.query(BacktestRun).filter(
|
|
BacktestRun.id == backtest_id
|
|
).first()
|
|
|
|
if not backtest_run:
|
|
return False
|
|
|
|
db_session.delete(backtest_run)
|
|
db_session.commit()
|
|
|
|
return True
|