Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -39,13 +39,13 @@ docker build --target test ./backtest_be_fast

**DB schema:** `database/schema.sql` is the first-boot initdb script; schema changes go through Alembic (`backtest_be_fast/alembic/`). There are **no physical foreign keys** — cross-table references are logical, integrity is owned by the app (A-10). A DB created from schema.sql **before** revision `7b2e9c4f1a30` still has the old FK: baseline it with `alembic stamp d5c3763b29e6` then `alembic upgrade head` (plain `stamp head` would skip the FK drop). A DB created from the current schema.sql is already at head (`alembic stamp head`). Migrations are NOT run by the GitHub Actions pipeline or the runtime image — apply them manually. schema.sql and the Alembic head must stay identical (COMMENTs included): change both together and run `scripts/check-schema-parity.sh`. Keep the `SET NAMES utf8mb4;` at the top of schema.sql — the official image runs initdb with a latin1 client and would double-encode Korean COMMENTs. Upgrade/baseline procedures and the MySQL 8.0 → 8.4 notes live in `database/README.md`.

**API:** POST `/api/v1/backtest` — main endpoint. Errors: `@handle_portfolio_errors` decorator.
**API:** POST `/api/v1/backtest` calculates results; GET `/api/v1/market-data` independently loads reference data. The console disables all POST `include_*` flags, renders the calculation first, then loads/retries reference data without rerunning it. Legacy POST defaults remain unchanged. See `docs/architecture/api-boundaries.md`. Errors: `@handle_portfolio_errors` decorator on the calculation path.

## Critical Constraints

1. **Async/Sync boundary:** Always `await asyncio.to_thread(sync_fn, ...)` for sync I/O in async endpoints. Violating this causes race conditions (first run fails, second succeeds).

2. **backtesting.py pinned to 0.3.3:** No `finalize_trades`, no `spread` param, commission on entry only, no Kelly Criterion.
2. **backtesting.py version is pinned in requirements.lock.txt** (currently 0.6.6). Preserve the tested commission/finalization behavior; do not apply assumptions from the historical 0.3.3 implementation.

3. **Strategy enum values:** `sma_strategy`, `rsi_strategy`, `bollinger_strategy`, `macd_strategy`, `buy_hold_strategy` (NOT "buy_and_hold"), `ema_strategy`. `PortfolioBacktestRequest.strategy` validates membership against `StrategyType` — unknown values are rejected, not silently accepted.

Expand Down
25 changes: 25 additions & 0 deletions HISTORY.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,31 @@

---

## 2026-10-25 API 경계·게이트웨이 보완

- **A-38**: 콘솔의 계산 POST와 참고 자료 GET을 분리했다. 기존 POST 본문·기본값·응답
정의는 보존하며, `include_*`를 모두 끈 계산에서는 메타데이터를 포함한 참고 조회를
생략한다. `GET /api/v1/market-data`는 별도 동시 실행 슬롯·대기·시간 한도로 동작한다.
- 참고 자료 지연·오류가 이미 완료한 계산의 표시나 입력 폼을 막지 않는다. 조회만
재시도하고 새 실행·화면 이탈 때 이전 요청을 취소한다. 공유 axios 전역 signal을
제거하고 요청별 signal을 명시적으로 전달하며 오래된 조회의 결과 덮어쓰기를 막는다.
- FE 포트폴리오 응답 타입을 실제 `PortfolioData`와 통일했다(A-24의 타입 불일치 부분).
종목별 응답 형태·지표 의미의 나머지 개선은 TODO에 유지한다.
- 운영 게이트웨이가 Cloudflare edge IP를 사용자로 취급하던 설정을 수정했다. 허용된
Cloudflare 대역의 `CF-Connecting-IP`만 신뢰하고 하위 XFF는 실제 사용자 주소로
덮어쓴다. 외부 XFF 위조 요청의 전후 로그로 edge IP → 실제 사용자 IP 전환과
위조 값 거부를 확인했다. 개인 IP 원문은 보호된 로컬 백업에만 둔다.
- 외부 게이트웨이의 기본 60초 읽기 제한을 180초로 명시해 앱의 최대 대기 30초 +
실행 60초와 맞췄다. Nginx 설정 검사 후 reload했고 실제 SSH·health·API를 확인한다.
- 기존 멀티프로세스 경합 테스트가 짧은 작업의 우연한 겹침에 의존해 peak=1로
실패할 수 있어, 첫 두 슬롯이 실제로 동시에 차도록 barrier를 추가했다.
- Docker에서 BE 512개·골든 E2E 1개, 콘솔 348개와 정적 검증을 통과했다.
배포용 이미지·격리 MySQL·합성 시세를 사용한 HTTP 및 브라우저 검증에서 기존
계산값 보존, 참고 GET 장애 중 결과 보존, 계산 재실행 없는 조회 복구를 확인했다.
- 구조와 호환성·검증 방법은 [API 경계](docs/architecture/api-boundaries.md)에 기록한다.

---

## 2026-10-10 감사 후속 수정

- **A-33**: 실패한 전략 종목의 투자금을 수익률 분모에서 제외하고 성공 항목의 비중을 다시 계산한다. 결과가 없거나 `final_equity`가 없는 종목도 경고에 포함한다. 남은 투자금이 0이면 명확한 입력 오류를 반환한다.
Expand Down
4 changes: 4 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,10 @@ CA 파일은 `/opt/backtest/production/tls/oracle-mysql-ca.pem`에 두고 컨테
## API와 프론트엔드 계약

`POST /api/v1/backtest`가 주요 API입니다. 최소 백테스트 기간은 30일입니다.

콘솔은 `include_*`를 끄고 계산 결과를 먼저 받은 뒤 `GET /api/v1/market-data`로
참고 데이터를 조회합니다. 기존 통합 응답은 기본 옵션으로 계속 지원합니다.
조회·취소·실행 한도와 이전 절차는 [API 경계](docs/architecture/api-boundaries.md)를 참고하세요.
프론트엔드의 `MIN_BACKTEST_PERIOD_DAYS`와 백엔드 설정을 함께 유지합니다.
CORS는 공통 Nginx에서 처리합니다. 서버 비밀값을 프론트엔드 `VITE_*` 변수에 넣지 않습니다.

Expand Down
4 changes: 4 additions & 0 deletions TODO.md
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,10 @@
다르다(가격 vs 금액). FE `api-types.ts`의 `PortfolioBacktestResponse`(`chart_data`, `stats` 등)는 실제
응답과 달라 화면이 `PortfolioData`로 캐스팅해 읽는다. BE 항목 형태를 하나로 정하고 FE 타입을 합친다.

2026-10-25: FE의 포트폴리오 응답 타입은 실제 `PortfolioData`로 통일하고 fixture의
이중 캐스팅을 제거했다. 계산/참고 조회 분리는 HISTORY의 A-38. 위 항목 중 남은
작업은 서버의 종목별 항목 키·필드 의미를 일관되게 정하는 것이다.

- [ ] **A-25 [be]** 뉴스 수집이 종목별로 순차 실행 〔배치9 api〕

종목당 10초 타임아웃, 20종목이면 최악 200초. 응답은 부가 수집 예산(15초)으로 막히고 남은 스레드는
Expand Down
9 changes: 6 additions & 3 deletions backtest_be_fast/app/api/v1/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,13 @@
- 백테스트 관련 모든 엔드포인트를 하나의 라우터로 구성

**엔드포인트**:
- POST /api/v1/backtest: 백테스트 실행 및 필요한 모든 데이터 응답
- POST /api/v1/backtest: 계산 실행 (기존 include_* 통합 응답도 지원)
- GET /api/v1/market-data: 계산과 독립적인 주가·환율·벤치마크·뉴스 조회
- 전략 목록은 프론트엔드에서 관리
- 주가/환율/뉴스 데이터는 백테스트 응답에 포함
- 콘솔은 계산을 먼저 표시하고 참고 데이터를 별도로 조회한다.
"""
from fastapi import APIRouter
from .endpoints import backtest
from .endpoints import backtest, market_data

api_router = APIRouter()

Expand All @@ -22,3 +23,5 @@
tags=["백테스팅"]
)

api_router.include_router(market_data.router, prefix="/market-data", tags=["시장 데이터"])

10 changes: 10 additions & 0 deletions backtest_be_fast/app/api/v1/endpoints/backtest.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,16 @@ async def _execute_portfolio_backtest(request: PortfolioBacktestRequest) -> dict
# 이 지점에 도달했다면 항상 성공 결과다.
backtest_result = await portfolio_manager_service.run_portfolio_backtest(request)

# The console loads reference data separately. Do not even query metadata
# when every supplemental section is disabled.
if not any((request.include_news, request.include_stock_data,
request.include_volatility_events, request.include_exchange_rates,
request.include_benchmarks)):
backtest_result['data'].update(
unified_data_service.empty_unified_data(symbols, outcome='skipped')
)
return backtest_result

# 3. 부가 데이터 수집 (데이터 서비스 위임) — asyncio.to_thread로 이벤트 루프 블로킹 방지
# [A-08] 요청의 include_* 옵션(기본값 모두 True = 기존 응답과 동일)으로 섹션을
# 끌 수 있고, 시간 예산을 넘긴 섹션은 비워서 반환한다. 부가 수집이 통째로
Expand Down
94 changes: 94 additions & 0 deletions backtest_be_fast/app/api/v1/endpoints/market_data.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
"""Reference market data, independently retriable without another simulation."""
import asyncio
from datetime import date
from functools import lru_cache
from typing import Annotated

from fastapi import APIRouter, Query, Request
from prometheus_client import Gauge
from pydantic import BaseModel, Field, field_validator, model_validator

from ....core.config import settings
from ....schemas.schemas import PortfolioStock
from ....services.backtest_runner import BacktestJobRunner, BacktestRejected
from ....services.unified_data_service import unified_data_service
from ....utils.serializers import recursive_serialize
from .backtest import _client_key, _rejection_response

router = APIRouter()
_running = Gauge('market_data_jobs_running', 'Active reference-data jobs', multiprocess_mode='livesum')
_waiting = Gauge('market_data_jobs_waiting', 'Queued reference-data jobs', multiprocess_mode='livesum')


class MarketDataQuery(BaseModel):
model_config = {'extra': 'forbid'}
symbols: list[str] = Field(default_factory=list, max_length=settings.max_portfolio_items)
start_date: date
end_date: date
include_stock_data: bool = True
include_volatility_events: bool = True
include_exchange_rates: bool = True
include_benchmarks: bool = True
include_news: bool = True

@field_validator('symbols')
@classmethod
def validate_symbols(cls, values):
return list(dict.fromkeys(PortfolioStock(symbol=value).symbol for value in values))

@model_validator(mode='after')
def validate_dates(self):
if self.end_date < self.start_date:
raise ValueError('종료일은 시작일 이후여야 합니다.')
if self.end_date > date.today():
raise ValueError('미래 날짜는 조회할 수 없습니다.')
if (self.end_date - self.start_date).days > settings.max_backtest_duration_days:
raise ValueError('조회 기간이 최대 허용 기간을 초과했습니다.')
return self


@lru_cache(maxsize=1)
def get_market_data_runner():
return BacktestJobRunner(
slot_dir=settings.backtest_slot_dir + '-market-data',
max_concurrent=settings.max_concurrent_market_data,
per_client=1,
queue_timeout=settings.market_data_queue_timeout_seconds,
exec_timeout=settings.market_data_timeout_seconds,
cancel_grace=settings.backtest_cancel_grace_seconds,
running_metric=_running,
waiting_metric=_waiting,
job_name='참고 데이터 조회',
)


@router.get('', summary='주가·환율·벤치마크·뉴스 조회')
async def get_market_data(query: Annotated[MarketDataQuery, Query()], request: Request):
async def collect():
data = await asyncio.to_thread(
unified_data_service.collect_all_unified_data,
symbols=query.symbols,
start_date=query.start_date.isoformat(),
end_date=query.end_date.isoformat(),
include_news=query.include_news,
news_display_count=15,
include_stock_data=query.include_stock_data,
include_volatility_events=query.include_volatility_events,
include_exchange_rates=query.include_exchange_rates,
include_benchmarks=query.include_benchmarks,
timeout_seconds=settings.supplemental_data_timeout_seconds,
)
benchmark = data.get('sp500_benchmark', [])
data['sp500_total_return_pct'] = (
unified_data_service.calculate_benchmark_return(benchmark)
if len(benchmark) >= 2 else None
)
return {'status': 'success', 'data': recursive_serialize(data)}

try:
return await get_market_data_runner().run(
collect, client_key=_client_key(request),
is_disconnected=request.is_disconnected, label='market-data',
)
except BacktestRejected as exc:
return _rejection_response(exc)
6 changes: 6 additions & 0 deletions backtest_be_fast/app/core/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,12 @@ def backend_cors_origins(self) -> list[str]:
# 이 안에 끝나지 않은 섹션은 비워서 반환하고 핵심 결과는 그대로 돌려준다.
# 로컬 실측(워밍 상태) supplemental_total p95 약 2초, 콜드 벤치마크 조회 약 5.5초.
supplemental_data_timeout_seconds: float = 15.0

# Read-only market data has its own capacity; slow feeds must not occupy
# the simulation slots or keep a completed calculation waiting.
max_concurrent_market_data: int = 2
market_data_queue_timeout_seconds: float = 2.0
market_data_timeout_seconds: float = 20.0

# pydantic v2 configuration
model_config = {
Expand Down
20 changes: 13 additions & 7 deletions backtest_be_fast/app/services/backtest_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,9 @@ def __init__(
cancel_grace: float,
poll_interval: float = 0.05,
disconnect_poll_interval: float = 1.0,
running_metric=BACKTEST_JOBS_RUNNING,
waiting_metric=BACKTEST_JOBS_WAITING,
job_name: str = "백테스트",
) -> None:
self.max_concurrent = max(1, int(max_concurrent))
self.per_client = int(per_client)
Expand All @@ -141,6 +144,9 @@ def __init__(
self.cancel_grace = float(cancel_grace)
self.poll_interval = float(poll_interval)
self.disconnect_poll_interval = float(disconnect_poll_interval)
self.running_metric = running_metric
self.waiting_metric = waiting_metric
self.job_name = job_name
self.slot_dir = slot_dir
self.global_pool = FileSlotPool(slot_dir, "global", self.max_concurrent)
self._client_pools: Dict[int, FileSlotPool] = {}
Expand All @@ -165,12 +171,12 @@ def active_jobs(self) -> int:
def _job_started(self) -> None:
with self._active_lock:
self._active += 1
BACKTEST_JOBS_RUNNING.inc()
self.running_metric.inc()

def _job_finished(self) -> None:
with self._active_lock:
self._active -= 1
BACKTEST_JOBS_RUNNING.dec()
self.running_metric.dec()

def shutdown(self, wait: bool = True) -> None:
self._executor.shutdown(wait=wait)
Expand Down Expand Up @@ -202,7 +208,7 @@ async def _wait_for_global_slot(
now = loop_time()
if now >= deadline:
raise QueueTimeout(
f"요청이 몰려 {self.queue_timeout:.0f}초 동안 기다렸지만 백테스트를 "
f"요청이 몰려 {self.queue_timeout:.0f}초 동안 기다렸지만 {self.job_name}를 "
f"시작하지 못했습니다(동시 실행 {self.max_concurrent}건 사용 중). "
"잠시 후 다시 시도해주세요.",
retry_after=max(1, int(self.exec_timeout // 2)),
Expand Down Expand Up @@ -235,14 +241,14 @@ async def run(
if client_slot is None:
logger.warning("백테스트 IP별 동시 실행 상한 초과: client=%s", client_key)
raise ClientLimitExceeded(
f"같은 클라이언트에서 이미 {self.per_client}건의 백테스트가 실행 중입니다. "
f"같은 클라이언트에서 이미 {self.per_client}건의 {self.job_name}가 실행 중입니다. "
"이전 요청이 끝난 뒤 다시 시도해주세요.",
retry_after=10,
)

wait_started = time.monotonic()
try:
with BACKTEST_JOBS_WAITING.track_inprogress():
with self.waiting_metric.track_inprogress():
global_slot = await self._wait_for_global_slot(is_disconnected)
except BaseException:
if client_slot is not None:
Expand Down Expand Up @@ -287,7 +293,7 @@ async def _wait_for_job(
if remaining <= 0:
self._cancel(token, "timeout", fut, job_id)
raise ExecutionTimeout(
f"백테스트 처리 시간이 {self.exec_timeout:.0f}초를 초과하여 중단되었습니다. "
f"{self.job_name} 처리 시간이 {self.exec_timeout:.0f}초를 초과하여 중단되었습니다. "
"기간을 줄이거나 잠시 후 다시 시도해주세요.",
retry_after=30,
)
Expand All @@ -300,7 +306,7 @@ async def _wait_for_job(
return
if is_disconnected is not None and await is_disconnected():
self._cancel(token, "client disconnected", fut, job_id)
raise ClientDisconnected("클라이언트가 연결을 끊어 백테스트를 중단했습니다.")
raise ClientDisconnected(f"클라이언트가 연결을 끊어 {self.job_name}를 중단했습니다.")

def _cancel(self, token: CancelToken, reason: str, fut: asyncio.Future, job_id: str) -> None:
if fut.done():
Expand Down
10 changes: 9 additions & 1 deletion backtest_be_fast/tests/unit/backtest_runner_mp_helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,8 @@
import time


def run_jobs_in_process(slot_dir: str, cap: int, jobs: int, job_seconds: float, log_path: str) -> None:
def run_jobs_in_process(slot_dir: str, cap: int, jobs: int, job_seconds: float, log_path: str,
first_wave, started) -> None:
"""이 프로세스 안에서 BacktestJobRunner로 작업 `jobs`개를 동시에 실행하고
각 작업의 [시작, 끝] 벽시계 시각을 log_path에 한 줄씩 남긴다."""
from app.services.backtest_runner import BacktestJobRunner
Expand All @@ -26,6 +27,13 @@ def run_jobs_in_process(slot_dir: str, cap: int, jobs: int, job_seconds: float,

def blocking_work() -> None:
start = time.time()
# Ensure actual contention independently of spawn/import/CPU scheduling.
# A wrongly serial runner fails the barrier rather than appearing healthy.
with started.get_lock():
started.value += 1
ordinal = started.value
if ordinal <= cap:
first_wave.wait(timeout=20)
time.sleep(job_seconds)
end = time.time()
line = json.dumps({"pid": os.getpid(), "start": start, "end": end}) + "\n"
Expand Down
4 changes: 4 additions & 0 deletions backtest_be_fast/tests/unit/test_api_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,5 +22,9 @@ def test_preserves_production_routes_requests_and_responses():
response = http.get("/api/v1/openapi.json")
assert response.status_code == 200
actual = response.json()
# The independently retriable read API is additive. Every pre-existing
# route and request/response schema must remain byte-for-byte equivalent.
assert set(actual['paths']) == set(expected['paths']) | {'/api/v1/market-data'}
assert 'get' in actual['paths'].pop('/api/v1/market-data')
for section in ("paths", "components"):
assert _contract(actual[section]) == expected[section]
Original file line number Diff line number Diff line change
Expand Up @@ -165,12 +165,14 @@ def test_three_processes_never_exceed_the_shared_cap(self, tmp_path):
log_path = str(tmp_path / "jobs.log")

ctx = multiprocessing.get_context("spawn")
first_wave = ctx.Barrier(cap)
started = ctx.Value('i', 0)
from tests.unit.backtest_runner_mp_helper import run_jobs_in_process

workers = [
ctx.Process(
target=run_jobs_in_process,
args=(slot_dir, cap, jobs_per_proc, job_seconds, log_path),
args=(slot_dir, cap, jobs_per_proc, job_seconds, log_path, first_wave, started),
)
for _ in range(procs)
]
Expand Down
Loading