AI 자동매매 매매 결정 엔드포인트 구현

.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

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")

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")

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

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"

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")

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=())

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(

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

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,
+    )

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