diff --git a/.env.example b/.env.example index e69de29..d0f6c75 100644 --- a/.env.example +++ b/.env.example @@ -0,0 +1,8 @@ +# Spring Boot <-> FastAPI 내부 호출 인증키 (JWT 아님) +INTERNAL_API_KEY=change-me + +# AI 서비스 전용 스키마. Spring Boot DB와 분리되어 있다. +DATABASE_URL=postgresql+psycopg://mlflow:mlflow@localhost:5432/arena_ai + +# 학습된 LightGBM 모델 경로. 없으면 스텁으로 기동한다. +MODEL_PATH=models/trading_lgbm.txt diff --git a/alembic/versions/20260810_0002_add_participant_and_strategy.py b/alembic/versions/20260810_0002_add_participant_and_strategy.py new file mode 100644 index 0000000..e9562fe --- /dev/null +++ b/alembic/versions/20260810_0002_add_participant_and_strategy.py @@ -0,0 +1,41 @@ +"""add participant_id and ai_strategy to decision_logs + +Revision ID: 20260810_0002 +Revises: 20260708_0001 +Create Date: 2026-08-10 +""" + +from typing import Sequence + +from alembic import op +import sqlalchemy as sa + +revision: str = "20260810_0002" +down_revision: str | None = "20260708_0001" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + # 0001과 같은 방식: native enum 대신 String + CHECK. + op.add_column( + "decision_logs", + sa.Column("participant_id", sa.BigInteger(), nullable=False, server_default="0"), + ) + op.add_column( + "decision_logs", + sa.Column("ai_strategy", sa.String(length=10), nullable=False, server_default="STABLE"), + ) + op.create_check_constraint( + "ck_decision_logs_ai_strategy", + "decision_logs", + "ai_strategy IN ('STABLE', 'AGGRESSIVE', 'TREND')", + ) + op.create_index("idx_decision_logs_participant_decided", "decision_logs", ["participant_id", "decided_at"]) + + +def downgrade() -> None: + op.drop_index("idx_decision_logs_participant_decided", table_name="decision_logs") + op.drop_constraint("ck_decision_logs_ai_strategy", "decision_logs", type_="check") + op.drop_column("decision_logs", "ai_strategy") + op.drop_column("decision_logs", "participant_id") diff --git a/app/main.py b/app/main.py index e12160d..5ca3a35 100644 --- a/app/main.py +++ b/app/main.py @@ -1,6 +1,22 @@ +from contextlib import asynccontextmanager + from fastapi import FastAPI -app = FastAPI() +from app.shared.http import register_exception_handlers +from app.trading_ai import predictor +from app.trading_ai.router import router as trading_ai_router + + +@asynccontextmanager +async def lifespan(app: FastAPI): + predictor.load_model() + yield + + +# ponytail: title은 건드리지 않는다. CI가 app.title == 'FastAPI'를 검사한다. +app = FastAPI(lifespan=lifespan) +register_exception_handlers(app) +app.include_router(trading_ai_router) @app.get("/health") diff --git a/app/observability/__init__.py b/app/observability/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/observability/decision_log.py b/app/observability/decision_log.py new file mode 100644 index 0000000..dbc88ae --- /dev/null +++ b/app/observability/decision_log.py @@ -0,0 +1,45 @@ +from datetime import datetime, timezone +from decimal import Decimal + +from sqlalchemy.orm import Session + +from app.shared.enums import AiStrategy, Market, TradingAction +from app.trading_ai.models import DecisionLog +from app.trading_ai.repositories import DecisionLogRepository + + +def record_decision( + db: Session, + *, + challenge_id: int, + participant_id: int, + symbol_code: str, + market: Market, + ai_strategy: AiStrategy, + features: dict[str, float], + probability: float, + action: TradingAction, + model_version: str, + order_id: int | None = None, +) -> int: + """결정 로그 한 행을 남기고 id를 돌려준다. 추천 엔드포인트도 같은 함수를 쓴다. + + SHAP 설명, 관측 대시보드, 재학습이 전부 이 행에 의존한다 (CLAUDE.md 비협상 항목). + """ + log = DecisionLog( + challenge_id=challenge_id, + participant_id=participant_id, + order_id=order_id, + symbol_code=symbol_code, + market=market, + ai_strategy=ai_strategy, + feature_snapshot=features, + model_output_probability=Decimal(str(round(probability, 4))), + action=action, + model_version=model_version, + decided_at=datetime.now(timezone.utc), + ) + DecisionLogRepository(db).add(log) # flush까지 하므로 여기서 id가 잡힌다 + decision_id = log.id + db.commit() + return decision_id diff --git a/app/shared/enums.py b/app/shared/enums.py index 76c7574..26a238e 100644 --- a/app/shared/enums.py +++ b/app/shared/enums.py @@ -13,6 +13,12 @@ class TradingAction(str, Enum): HOLD = "HOLD" +class AiStrategy(str, Enum): + STABLE = "STABLE" + AGGRESSIVE = "AGGRESSIVE" + TREND = "TREND" + + class ModelStatus(str, Enum): CHAMPION = "CHAMPION" CHALLENGER = "CHALLENGER" diff --git a/app/shared/http.py b/app/shared/http.py new file mode 100644 index 0000000..858b7fa --- /dev/null +++ b/app/shared/http.py @@ -0,0 +1,45 @@ +import logging +import os +import secrets +from http import HTTPStatus + +from fastapi import FastAPI, Header, HTTPException, Request, status +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse + +logger = logging.getLogger(__name__) + + +def verify_internal_api_key(x_internal_api_key: str | None = Header(default=None)) -> None: + """Spring Boot -> AI 서비스 내부 호출 인증. JWT 아님, 라우터에 Depends로 붙인다.""" + expected = os.getenv("INTERNAL_API_KEY") + if not expected: + raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "INTERNAL_API_KEY is not configured") + if x_internal_api_key is None or not secrets.compare_digest(x_internal_api_key, expected): + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid internal api key") + + +def error_response(status_code: int, message: str) -> JSONResponse: + try: + code = HTTPStatus(status_code).name + except ValueError: + code = "ERROR" + return JSONResponse(status_code=status_code, content={"error": {"code": code, "message": message}}) + + +def register_exception_handlers(app: FastAPI) -> None: + # ponytail: 성공 응답은 감싸지 않는다 (명세가 bare object). 에러 응답만 포맷을 통일한다. + @app.exception_handler(HTTPException) + async def _http_exception(request: Request, exc: HTTPException) -> JSONResponse: + return error_response(exc.status_code, str(exc.detail)) + + @app.exception_handler(RequestValidationError) + async def _validation_error(request: Request, exc: RequestValidationError) -> JSONResponse: + message = "; ".join(f"{'.'.join(map(str, e['loc']))}: {e['msg']}" for e in exc.errors()) + return error_response(status.HTTP_422_UNPROCESSABLE_CONTENT, message) + + @app.exception_handler(Exception) + async def _unhandled(request: Request, exc: Exception) -> JSONResponse: + # 명세의 "500 - 모델 추론 실패"가 여기로 떨어진다. 스택트레이스는 로그로만, 응답에는 싣지 않는다. + logger.exception("unhandled error on %s %s", request.method, request.url.path) + return error_response(status.HTTP_500_INTERNAL_SERVER_ERROR, "internal server error") diff --git a/app/shared/schemas.py b/app/shared/schemas.py new file mode 100644 index 0000000..1027b2c --- /dev/null +++ b/app/shared/schemas.py @@ -0,0 +1,9 @@ +from pydantic import BaseModel, ConfigDict +from pydantic.alias_generators import to_camel + + +class CamelModel(BaseModel): + """요청/응답 JSON은 camelCase, 파이썬 속성은 snake_case.""" + + # protected_namespaces=(): modelVersion 같은 필드가 pydantic의 model_ 예약 접두사와 부딪히는 것을 막는다. + model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True, protected_namespaces=()) diff --git a/app/trading_ai/models.py b/app/trading_ai/models.py index f31a526..ab22d7b 100644 --- a/app/trading_ai/models.py +++ b/app/trading_ai/models.py @@ -7,24 +7,30 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship from app.shared.db import Base -from app.shared.enums import Market, TradingAction +from app.shared.enums import AiStrategy, Market, TradingAction class DecisionLog(Base): __tablename__ = "decision_logs" __table_args__ = ( Index("idx_decision_logs_challenge_decided", "challenge_id", "decided_at"), + Index("idx_decision_logs_participant_decided", "participant_id", "decided_at"), Index("idx_decision_logs_model_version", "model_version"), ) id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) challenge_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + participant_id: Mapped[int] = mapped_column(BigInteger, nullable=False) order_id: Mapped[int | None] = mapped_column(BigInteger) symbol_code: Mapped[str] = mapped_column(String(20), nullable=False) market: Mapped[Market] = mapped_column( Enum(Market, native_enum=False, name="ck_decision_logs_market"), nullable=False, ) + ai_strategy: Mapped[AiStrategy] = mapped_column( + Enum(AiStrategy, native_enum=False, name="ck_decision_logs_ai_strategy"), + nullable=False, + ) feature_snapshot: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False) model_output_probability: Mapped[Decimal] = mapped_column(Numeric(5, 4), nullable=False) action: Mapped[TradingAction] = mapped_column( diff --git a/app/trading_ai/predictor.py b/app/trading_ai/predictor.py new file mode 100644 index 0000000..6a4ae27 --- /dev/null +++ b/app/trading_ai/predictor.py @@ -0,0 +1,70 @@ +import logging +import os +from pathlib import Path + +from app.shared.enums import AiStrategy, Market, TradingAction + +logger = logging.getLogger(__name__) + +STUB_VERSION = "stub-0" + +# 시장 x 전략별 (buy_above, sell_below). 사이는 HOLD. +# COIN은 변동성이 커서 US와 같은 확률에 같은 결정을 내리면 안 된다 -> HOLD 구간을 넓게 잡는다. +# ponytail: 백테스트로 튜닝해야 하는 값이라 상수로 노출해 둔다. 결과 나오면 여기만 고친다. +THRESHOLDS: dict[Market, dict[AiStrategy, tuple[float, float]]] = { + Market.KR: { + AiStrategy.STABLE: (0.70, 0.30), + AiStrategy.AGGRESSIVE: (0.55, 0.45), + AiStrategy.TREND: (0.62, 0.38), + }, + Market.US: { + AiStrategy.STABLE: (0.70, 0.30), + AiStrategy.AGGRESSIVE: (0.55, 0.45), + AiStrategy.TREND: (0.62, 0.38), + }, + Market.COIN: { + AiStrategy.STABLE: (0.75, 0.25), + AiStrategy.AGGRESSIVE: (0.60, 0.40), + AiStrategy.TREND: (0.68, 0.32), + }, +} + +_booster = None +_version = STUB_VERSION + + +def load_model() -> None: + """lifespan에서 한 번 호출. 모델 파일이 없으면 스텁으로 남는다.""" + global _booster, _version + + path = Path(os.getenv("MODEL_PATH", "models/trading_lgbm.txt")) + if not path.exists(): + logger.warning("모델 파일 없음 (%s) - 스텁으로 기동한다", path) + return + + import lightgbm as lgb # 무거운 import라 실제로 쓸 때만 끌어온다 + + _booster = lgb.Booster(model_file=str(path)) + _version = path.stem + logger.info("모델 로드 완료: %s", _version) + + +def predict(features: dict[str, float]) -> tuple[float, str]: + """(상승 확률, 모델 버전).""" + if _booster is None: + # ponytail: 스텁. 학습된 Booster가 생기면 load_model()이 채우고 이 분기는 안 탄다. + return 0.5, STUB_VERSION + + import pandas as pd + + probability = float(_booster.predict(pd.DataFrame([features]))[0]) + return probability, _version + + +def decide_action(probability: float, market: Market, strategy: AiStrategy) -> TradingAction: + buy_above, sell_below = THRESHOLDS[market][strategy] + if probability >= buy_above: + return TradingAction.BUY + if probability <= sell_below: + return TradingAction.SELL + return TradingAction.HOLD diff --git a/app/trading_ai/router.py b/app/trading_ai/router.py new file mode 100644 index 0000000..7089f33 --- /dev/null +++ b/app/trading_ai/router.py @@ -0,0 +1,40 @@ +from fastapi import APIRouter, Depends +from sqlalchemy.orm import Session + +from app.observability.decision_log import record_decision +from app.shared.db import get_db +from app.shared.http import verify_internal_api_key +from app.trading_ai import predictor +from app.trading_ai.schemas import TradingDecisionRequest, TradingDecisionResponse + +router = APIRouter(prefix="/internal/ai", dependencies=[Depends(verify_internal_api_key)]) + + +@router.post("/trading-decisions", response_model=TradingDecisionResponse) +def create_trading_decision( + body: TradingDecisionRequest, + db: Session = Depends(get_db), +) -> TradingDecisionResponse: + probability, model_version = predictor.predict(body.features) + action = predictor.decide_action(probability, body.market, body.ai_strategy) + + # 로깅은 이 요청의 일부다. 나중에 따로 하는 것이 아니다. + decision_id = record_decision( + db, + challenge_id=body.challenge_id, + participant_id=body.participant_id, + symbol_code=body.symbol, + market=body.market, + ai_strategy=body.ai_strategy, + features=body.features, + probability=probability, + action=action, + model_version=model_version, + ) + + return TradingDecisionResponse( + decision_id=decision_id, + action=action, + probability=probability, + model_version=model_version, + ) diff --git a/app/trading_ai/schemas.py b/app/trading_ai/schemas.py new file mode 100644 index 0000000..e3ea43a --- /dev/null +++ b/app/trading_ai/schemas.py @@ -0,0 +1,19 @@ +from app.shared.enums import AiStrategy, Market, TradingAction +from app.shared.schemas import CamelModel + + +class TradingDecisionRequest(CamelModel): + challenge_id: int + participant_id: int + symbol: str + market: Market + ai_strategy: AiStrategy + # ponytail: 개별 피처 키는 검증하지 않는다. 모델 피처 목록이 확정되면 여기에 박는다. + features: dict[str, float] + + +class TradingDecisionResponse(CamelModel): + decision_id: int + action: TradingAction + probability: float + model_version: str diff --git a/tests/test_trading_decisions.py b/tests/test_trading_decisions.py new file mode 100644 index 0000000..60d2818 --- /dev/null +++ b/tests/test_trading_decisions.py @@ -0,0 +1,87 @@ +import os +import unittest +from unittest.mock import patch + +from fastapi.testclient import TestClient + +from app.main import app +from app.shared.db import get_db +from app.shared.enums import AiStrategy, Market, TradingAction +from app.trading_ai.predictor import decide_action + +API_KEY = "test-key" +URL = "/internal/ai/trading-decisions" +BODY = { + "challengeId": 1, + "participantId": 7, + "symbol": "AAPL", + "market": "US", + "aiStrategy": "STABLE", + "features": {"rsi": 55.0, "ma5": 1.0, "ma20": 1.0, "volumeChange": 0.1, "macd": 0.2, "volatility": 0.3}, +} + + +class TradingDecisionEndpointTest(unittest.TestCase): + def setUp(self): + os.environ["INTERNAL_API_KEY"] = API_KEY + # 실 DB를 띄우지 않는다. 로깅 호출 여부만 확인한다. + app.dependency_overrides[get_db] = lambda: None + self.client = TestClient(app, raise_server_exceptions=False) + + def tearDown(self): + app.dependency_overrides.clear() + os.environ.pop("INTERNAL_API_KEY", None) + + def test_missing_api_key_is_unauthorized(self): + response = self.client.post(URL, json=BODY) + self.assertEqual(response.status_code, 401) + self.assertEqual(response.json()["error"]["code"], "UNAUTHORIZED") + + def test_wrong_api_key_is_unauthorized(self): + response = self.client.post(URL, json=BODY, headers={"X-Internal-Api-Key": "nope"}) + self.assertEqual(response.status_code, 401) + + def test_valid_request_returns_camel_case_body(self): + with patch("app.trading_ai.router.record_decision", return_value=123) as record: + response = self.client.post(URL, json=BODY, headers={"X-Internal-Api-Key": API_KEY}) + + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.json(), + {"decisionId": 123, "action": "HOLD", "probability": 0.5, "modelVersion": "stub-0"}, + ) + # 결정 로깅은 엔드포인트의 일부다 (CLAUDE.md 비협상 항목). + record.assert_called_once() + kwargs = record.call_args.kwargs + self.assertEqual(kwargs["challenge_id"], 1) + self.assertEqual(kwargs["participant_id"], 7) + self.assertEqual(kwargs["ai_strategy"], AiStrategy.STABLE) + self.assertEqual(kwargs["features"], BODY["features"]) + + def test_invalid_body_returns_formatted_422(self): + response = self.client.post(URL, json={"challengeId": 1}, headers={"X-Internal-Api-Key": API_KEY}) + self.assertEqual(response.status_code, 422) + self.assertIn("error", response.json()) + + def test_inference_failure_returns_500(self): + with patch("app.trading_ai.predictor.predict", side_effect=RuntimeError("boom")): + response = self.client.post(URL, json=BODY, headers={"X-Internal-Api-Key": API_KEY}) + + self.assertEqual(response.status_code, 500) + self.assertEqual(response.json()["error"]["message"], "internal server error") + + +class DecideActionTest(unittest.TestCase): + def test_thresholds_are_separated_per_market(self): + # COIN은 변동성이 커서 US와 같은 확률에 같은 결정이 나오면 안 된다. + self.assertEqual(decide_action(0.72, Market.US, AiStrategy.STABLE), TradingAction.BUY) + self.assertEqual(decide_action(0.72, Market.COIN, AiStrategy.STABLE), TradingAction.HOLD) + + def test_stub_probability_is_hold_everywhere(self): + for market in Market: + for strategy in AiStrategy: + self.assertEqual(decide_action(0.5, market, strategy), TradingAction.HOLD) + + +if __name__ == "__main__": + unittest.main()