feat: implement User Data Stream for real-time TP/SL detection and PnL tracking

- Introduced User Data Stream to detect TP/SL executions in real-time.
- Added a new class `UserDataStream` for managing the stream and handling events.
- Updated `bot.py` to initialize and run the User Data Stream in parallel with the candle stream.
- Enhanced `notifier.py` to send detailed Discord notifications including estimated vs actual PnL.
- Added methods in `exchange.py` for managing listenKey lifecycle (create, keepalive, delete).
- Refactored PnL recording and notification logic to streamline handling of position closures.

Made-with: Cursor

.gitignore

@@ -7,5 +7,9 @@ logs/
 .venv/
 venv/
 models/*.pkl
+models/*.onnx
+models/tune_results_*.json
 data/*.parquet
 .worktrees/
+.DS_Store
+.cursor/

README.md

@@ -1,18 +1,30 @@
 # CoinTrader
 
-Binance Futures 자동매매 봇. 복합 기술 지표와 LightGBM ML 필터를 결합하여 XRPUSDT(기본) 선물 포지션을 자동으로 진입·청산하며, Discord로 실시간 알림을 전송합니다.
+Binance Futures 자동매매 봇. 복합 기술 지표와 ML 필터(LightGBM / MLX 신경망)를 결합하여 XRPUSDT(기본) 선물 포지션을 자동으로 진입·청산하며, Discord로 실시간 알림을 전송합니다.
 
 ---
 
 ## 주요 기능
 
-- **복합 기술 지표 신호**: RSI, MACD 크로스, 볼린저 밴드, EMA 정/역배열, Stochastic RSI, 거래량 급증 — 3개 이상 일치 시 진입
-- **ML 필터 (LightGBM)**: 기술 지표 신호를 한 번 더 검증하여 오진입 차단 (모델 없으면 자동 폴백)
+- **복합 기술 지표 신호**: RSI, MACD 크로스, 볼린저 밴드, EMA 정/역배열, Stochastic RSI, 거래량 급증 — 가중치 합계 ≥ 3 시 진입
+- **ML 필터 (ONNX 우선 / LightGBM 폴백)**: 기술 지표 신호를 한 번 더 검증하여 오진입 차단. 우선순위: ONNX > LightGBM > 폴백(항상 허용)
+- **모델 핫리로드**: 캔들마다 모델 파일 mtime을 감지해 변경 시 자동 리로드 (봇 재시작 불필요)
+- **멀티심볼 스트림**: XRP/BTC/ETH 3개 심볼을 단일 Combined WebSocket으로 수신, BTC·ETH 상관관계 피처 활용
+- **23개 ML 피처**: XRP 기술 지표 13개 + BTC/ETH 수익률·상대강도 8개 + OI 변화율·펀딩비 2개 (캔들 마감 시 실시간 조회, 실패 시 0으로 폴백)
+- **점진적 OI 데이터 축적 (Upsert)**: 바이낸스 OI 히스토리 API는 최근 30일치만 제공. `fetch_history.py` 실행 시 기존 parquet의 `oi_change/funding_rate=0` 구간을 신규 값으로 채워 학습 데이터 품질을 점진적으로 개선
+- **실시간 OI/펀딩비 조회**: 캔들 마감마다 `get_open_interest()` / `get_funding_rate()`를 비동기 병렬 조회하여 ML 피처에 전달. 이전 캔들 대비 OI 변화율로 변환하여 train-serve skew 해소
 - **ATR 기반 손절/익절**: 변동성에 따라 동적으로 SL/TP 계산 (1.5× / 3.0× ATR)
-- **리스크 관리**: 트레이드당 리스크 비율, 최대 포지션 수, 일일 손실 한도 제어
+- **Algo Order API 지원**: 계정 설정에 따라 STOP_MARKET/TAKE_PROFIT_MARKET 주문을 `/fapi/v1/algoOrder` 엔드포인트로 자동 전송 (오류 코드 -4120 대응)
+- **동적 증거금 비율**: 잔고 증가에 따라 선형 감소 (최대 50% → 최소 20%)
+- **반대 시그널 재진입**: 보유 포지션과 반대 신호 발생 시 즉시 청산 후 ML 필터 통과 시 반대 방향 재진입
+- **리스크 관리**: 트레이드당 리스크 비율, 최대 포지션 수, 일일 손실 한도(5%) 제어
 - **포지션 복구**: 봇 재시작 시 기존 포지션 자동 감지 및 상태 복원
+- **실시간 TP/SL 감지**: Binance User Data Stream으로 TP/SL 작동을 즉시 감지 (캔들 마감 대기 없음)
+- **순수익(Net PnL) 기록**: 바이낸스 `realizedProfit - commission`으로 정확한 순수익 계산
+- **Discord 상세 청산 알림**: 예상 수익 vs 실제 순수익 + 슬리피지/수수료 차이 표시
+- **listenKey 자동 갱신**: 30분 keepalive + 네트워크 단절 시 자동 재연결
 - **Discord 알림**: 진입·청산·오류 이벤트 실시간 웹훅 알림
-- **CI/CD**: Jenkins + Gitea Container Registry 기반 Docker 이미지 자동 빌드·배포
+- **CI/CD**: Jenkins + Gitea Container Registry 기반 Docker 이미지 자동 빌드·배포 (LXC 운영 서버 자동 적용)
 
 ---
 
@@ -20,27 +32,34 @@ Binance Futures 자동매매 봇. 복합 기술 지표와 LightGBM ML 필터를
 
 ```
 cointrader/
-├── main.py                  # 진입점
+├── main.py                    # 진입점
 ├── src/
-│   ├── bot.py               # 메인 트레이딩 루프
-│   ├── config.py            # 환경변수 기반 설정
-│   ├── exchange.py          # Binance Futures API 클라이언트
-│   ├── data_stream.py       # WebSocket 1분봉 스트림
-│   ├── indicators.py        # 기술 지표 계산 및 신호 생성
-│   ├── ml_filter.py         # LightGBM 진입 필터
-│   ├── ml_features.py       # ML 피처 빌더
-│   ├── label_builder.py     # 학습 레이블 생성
-│   ├── dataset_builder.py   # 벡터화 데이터셋 빌더 (학습용)
-│   ├── risk_manager.py      # 리스크 관리
-│   ├── notifier.py          # Discord 웹훅 알림
-│   └── logger_setup.py      # Loguru 로거 설정
+│   ├── bot.py                 # 메인 트레이딩 루프
+│   ├── config.py              # 환경변수 기반 설정
+│   ├── exchange.py            # Binance Futures API 클라이언트
+│   ├── data_stream.py         # WebSocket 15분봉 멀티심볼 스트림 (XRP/BTC/ETH)
+│   ├── indicators.py          # 기술 지표 계산 및 신호 생성
+│   ├── ml_filter.py           # ML 필터 (ONNX 우선 / LightGBM 폴백 / 핫리로드)
+│   ├── ml_features.py         # ML 피처 빌더 (23개 피처)
+│   ├── mlx_filter.py          # MLX 신경망 필터 (Apple Silicon GPU 학습 + ONNX export)
+│   ├── label_builder.py       # 학습 레이블 생성
+│   ├── dataset_builder.py     # 벡터화 데이터셋 빌더 (학습용)
+│   ├── risk_manager.py        # 리스크 관리 (일일 손실 한도, 동적 증거금 비율)
+│   ├── notifier.py            # Discord 웹훅 알림
+│   └── logger_setup.py        # Loguru 로거 설정
 ├── scripts/
-│   ├── fetch_history.py     # 과거 데이터 수집
-│   └── train_model.py       # ML 모델 수동 학습
-├── models/                  # 학습된 모델 저장 (.pkl)
-├── data/                    # 과거 데이터 캐시
-├── logs/                    # 로그 파일
-├── tests/                   # 테스트 코드
+│   ├── fetch_history.py       # 과거 데이터 수집 (XRP/BTC/ETH + OI/펀딩비, Upsert 지원)
+│   ├── train_model.py         # LightGBM 모델 학습 (CPU)
+│   ├── train_mlx_model.py     # MLX 신경망 학습 (Apple Silicon GPU)
+│   ├── train_and_deploy.sh    # 전체 파이프라인 (수집 → 학습 → LXC 배포)
+│   ├── tune_hyperparams.py    # Optuna 하이퍼파라미터 자동 탐색 (수동 트리거)
+│   ├── deploy_model.sh        # 모델 파일 LXC 서버 전송
+│   └── run_tests.sh           # 전체 테스트 실행
+├── models/                    # 학습된 모델 저장 (.pkl / .onnx)
+├── data/                      # 과거 데이터 캐시 (.parquet)
+├── logs/                      # 로그 파일
+├── docs/plans/                # 설계 문서 및 구현 플랜
+├── tests/                     # 테스트 코드
 ├── Dockerfile
 ├── docker-compose.yml
 ├── Jenkinsfile
@@ -64,7 +83,6 @@ BINANCE_API_KEY=your_api_key
 BINANCE_API_SECRET=your_api_secret
 SYMBOL=XRPUSDT
 LEVERAGE=10
-RISK_PER_TRADE=0.02
 DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/...
 ```
 
@@ -91,17 +109,82 @@ docker compose logs -f cointrader
 
 ## ML 모델 학습
 
-봇은 모델 파일(`models/lgbm_filter.pkl`)이 없으면 ML 필터 없이 동작합니다. 최초 실행 전 또는 수동 재학습 시 아래 순서로 진행합니다.
+봇은 모델 파일이 없으면 ML 필터 없이 동작합니다. 최초 실행 전 또는 수동 재학습 시 아래 순서로 진행합니다.
+
+### 전체 파이프라인 (권장)
+
+맥미니에서 데이터 수집 → 학습 → LXC 배포까지 한 번에 실행합니다.
+
+> **자동 분기**: `data/combined_15m.parquet`가 없으면 1년치(365일) 전체 수집, 있으면 35일치 Upsert로 자동 전환합니다. 서버 이전이나 데이터 유실 시에도 사람의 개입 없이 자동 복구됩니다.
 
 ```bash
-# 1. 과거 데이터 수집
-python scripts/fetch_history.py
+# LightGBM + Walk-Forward 5폴드 (기본값)
+bash scripts/train_and_deploy.sh
 
-# 2. 모델 학습 (LightGBM, CPU)
-python scripts/train_model.py
+# MLX GPU 학습 + Walk-Forward 5폴드
+bash scripts/train_and_deploy.sh mlx
+
+# LightGBM + Walk-Forward 3폴드
+bash scripts/train_and_deploy.sh lgbm 3
+
+# 학습만 (배포 없이)
+bash scripts/train_and_deploy.sh lgbm 0
 ```
 
-학습된 모델은 `models/lgbm_filter.pkl`에 저장됩니다. 재학습이 필요하면 맥미니에서 위 스크립트를 다시 실행하고 모델 파일을 컨테이너에 배포합니다.
+### 단계별 수동 실행
+
+```bash
+# 1. 과거 데이터 수집 (XRP/BTC/ETH 3심볼, 15분봉, 1년치 + OI/펀딩비)
+# 기본값: Upsert 활성화 — 기존 parquet의 oi_change/funding_rate=0 구간을 실제 값으로 채움
+python scripts/fetch_history.py \
+    --symbols XRPUSDT BTCUSDT ETHUSDT \
+    --interval 15m \
+    --days 365 \
+    --output data/combined_15m.parquet
+
+# 기존 파일을 완전히 덮어쓰려면 --no-upsert 플래그 사용
+python scripts/fetch_history.py \
+    --symbols XRPUSDT BTCUSDT ETHUSDT \
+    --interval 15m \
+    --days 365 \
+    --output data/combined_15m.parquet \
+    --no-upsert
+
+# 2-A. LightGBM 모델 학습 (CPU)
+python scripts/train_model.py --data data/combined_15m.parquet
+
+# 2-B. MLX 신경망 학습 (Apple Silicon GPU)
+python scripts/train_mlx_model.py --data data/combined_15m.parquet
+
+# 3. LXC 서버에 모델 배포
+bash scripts/deploy_model.sh        # LightGBM
+bash scripts/deploy_model.sh mlx    # MLX (ONNX)
+```
+
+학습된 모델은 `models/lgbm_filter.pkl` (LightGBM) 또는 `models/mlx_filter.weights.onnx` (MLX) 에 저장됩니다.
+
+> **모델 핫리로드**: 봇이 실행 중일 때 모델 파일을 교체하면, 다음 캔들 마감 시 자동으로 감지해 리로드합니다. 봇 재시작이 필요 없습니다.
+
+### 하이퍼파라미터 자동 튜닝 (Optuna)
+
+봇 성능이 저하되거나 데이터가 충분히 축적되었을 때 Optuna로 최적 LightGBM 파라미터를 탐색합니다.
+결과를 확인하고 직접 승인한 후 재학습에 반영하는 **수동 트리거** 방식입니다.
+
+```bash
+# 기본 실행 (50 trials, 5폴드 Walk-Forward, ~30분)
+python scripts/tune_hyperparams.py
+
+# 빠른 테스트 (10 trials, 3폴드, ~5분)
+python scripts/tune_hyperparams.py --trials 10 --folds 3
+
+# 베이스라인 측정 없이 탐색만
+python scripts/tune_hyperparams.py --no-baseline
+```
+
+결과는 `models/tune_results_YYYYMMDD_HHMMSS.json`에 저장됩니다.
+콘솔에 Best Params, 베이스라인 대비 개선폭, 폴드별 AUC를 출력하므로 직접 확인 후 판단하세요.
+
+> **주의**: Optuna가 찾은 파라미터는 과적합 위험이 있습니다. Best Params를 `train_model.py`에 반영하기 전에 반드시 폴드별 AUC 분산과 개선폭을 검토하세요.
 
 ### Apple Silicon GPU 가속 학습 (M1/M2/M3/M4)
 
@@ -110,23 +193,16 @@ M 시리즈 맥에서는 MLX를 사용해 통합 GPU(Metal)로 학습할 수 있
 > **설치**: `mlx`는 Apple Silicon 전용이며 `requirements.txt`에 포함되지 않습니다.
 > 맥미니에서 별도 설치: `pip install mlx`
 
-```bash
-# MLX 별도 설치 (맥미니 전용)
-pip install mlx
+MLX로 학습한 모델은 ONNX 포맷으로 export되어 Linux 서버에서 `onnxruntime`으로 추론합니다.
 
-# MLX 신경망 필터 학습 (GPU 자동 사용)
-python scripts/train_mlx_model.py
-
-# train_and_deploy.sh에서 MLX 백엔드 사용
-TRAIN_BACKEND=mlx bash scripts/train_and_deploy.sh
-```
-
-> **참고**: LightGBM은 Apple Silicon GPU를 공식 지원하지 않습니다. MLX는 Apple이 만든 ML 프레임워크로 통합 GPU를 자동으로 활용합니다. Neural Engine(NPU)은 Apple 내부 전용으로 Python에서 직접 제어할 수 없습니다.
+> **참고**: LightGBM은 Apple Silicon GPU를 공식 지원하지 않습니다. MLX는 Apple이 만든 ML 프레임워크로 통합 GPU를 자동으로 활용합니다.
 
 ---
 
 ## 매매 전략
 
+### 기술 지표 신호 (15분봉)
+
 | 지표 | 롱 조건 | 숏 조건 | 가중치 |
 |------|---------|---------|--------|
 | RSI (14) | < 35 | > 65 | 1 |
@@ -138,7 +214,13 @@ TRAIN_BACKEND=mlx bash scripts/train_and_deploy.sh
 
 **진입 조건**: 가중치 합계 ≥ 3 + (거래량 급증 또는 가중치 합계 ≥ 4)  
 **손절/익절**: ATR × 1.5 / ATR × 3.0 (리스크:리워드 = 1:2)  
-**ML 필터**: LightGBM 예측 확률 ≥ 0.60 이어야 최종 진입
+**ML 필터**: 예측 확률 ≥ 0.60 이어야 최종 진입
+
+### 반대 시그널 재진입
+
+보유 포지션과 반대 방향 신호가 발생하면:
+1. 기존 포지션 즉시 청산 (미체결 SL/TP 주문 취소 포함)
+2. ML 필터 통과 시 반대 방향으로 즉시 재진입
 
 ---
 
@@ -146,22 +228,27 @@ TRAIN_BACKEND=mlx bash scripts/train_and_deploy.sh
 
 `main` 브랜치에 푸시하면 Jenkins 파이프라인이 자동으로 실행됩니다.
 
-1. **Checkout** — 소스 체크아웃
-2. **Build Image** — Docker 이미지 빌드 (`:{BUILD_NUMBER}` + `:latest` 태그)
-3. **Push** — Gitea Container Registry(`10.1.10.28:3000`)에 푸시
-4. **Cleanup** — 로컬 이미지 정리
+1. **Notify Build Start** — Discord 빌드 시작 알림
+2. **Git Clone from Gitea** — 소스 체크아웃
+3. **Build Docker Image** — Docker 이미지 빌드 (`:{BUILD_NUMBER}` + `:latest` 태그)
+4. **Push to Gitea Registry** — Gitea Container Registry(`10.1.10.28:3000`)에 푸시
+5. **Deploy to Prod LXC** — 운영 LXC 서버(`10.1.10.24`)에 자동 배포 (`docker compose pull && up -d`)
+6. **Cleanup** — 빌드 서버 로컬 이미지 정리
 
-배포 서버에서 최신 이미지를 반영하려면:
-
-```bash
-docker compose pull && docker compose up -d
-```
+빌드 성공/실패 결과는 Discord로 자동 알림됩니다.
 
 ---
 
 ## 테스트
 
 ```bash
+# 전체 테스트
+bash scripts/run_tests.sh
+
+# 특정 키워드 필터
+bash scripts/run_tests.sh -k bot
+
+# pytest 직접 실행
 pytest tests/ -v
 ```
 
@@ -175,13 +262,15 @@ pytest tests/ -v
 | `BINANCE_API_SECRET` | — | Binance API 시크릿 |
 | `SYMBOL` | `XRPUSDT` | 거래 심볼 |
 | `LEVERAGE` | `10` | 레버리지 배수 |
-| `RISK_PER_TRADE` | `0.02` | 트레이드당 리스크 비율 (2%) |
 | `DISCORD_WEBHOOK_URL` | — | Discord 웹훅 URL |
+| `MARGIN_MAX_RATIO` | `0.50` | 최대 증거금 비율 (잔고 대비 50%) |
+| `MARGIN_MIN_RATIO` | `0.20` | 최소 증거금 비율 (잔고 대비 20%) |
+| `MARGIN_DECAY_RATE` | `0.0006` | 잔고 증가 시 증거금 비율 감소 속도 |
 
 ---
 
 ## 주의사항
 
 > **이 봇은 실제 자산을 거래합니다.** 운영 전 반드시 Binance Testnet에서 충분히 검증하세요.  
-> 과거 수익이 미래 수익을 보장하지 않습니다. 투자 손실에 대한 책임은 사용자 본인에게 있습니다.
+> 과거 수익이 미래 수익을 보장하지 않습니다. 투자 손실에 대한 책임은 사용자 본인에게 있습니다.  
 > 성투기원합니다.

docs/plans/2026-03-02-oi-funding-accumulation.md

@@ -0,0 +1,394 @@
+# OI/펀딩비 누적 저장 (접근법 B) 구현 계획
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** `fetch_history.py`의 데이터 수집 방식을 덮어쓰기(Overwrite)에서 Upsert(병합)로 변경해, 매일 실행할 때마다 기존 parquet의 OI/펀딩비 0.0 구간이 실제 값으로 채워지며 고품질 데이터가 무한히 누적되도록 한다.
+
+**Architecture:**
+- `fetch_history.py`에 `--upsert` 플래그 추가 (기본값 True). 기존 parquet이 있으면 로드 후 신규 데이터와 timestamp 기준 병합(Upsert). 없으면 기존처럼 새로 생성.
+- Upsert 규칙: 기존 행의 `oi_change` / `funding_rate`가 0.0이면 신규 값으로 덮어씀. 신규 행은 그냥 추가. 중복 제거 후 시간순 정렬.
+- `train_and_deploy.sh`의 `--days` 인자를 35일로 조정 (30일 API 한도 + 5일 버퍼).
+- LXC 운영서버는 모델 파일만 받으므로 변경 없음. 맥미니의 `data/` 폴더에만 누적.
+
+**Tech Stack:** pandas, parquet (pyarrow), pytest
+
+---
+
+## Task 1: fetch_history.py — upsert_parquet() 함수 추가 및 --upsert 플래그
+
+**Files:**
+- Modify: `scripts/fetch_history.py`
+- Test: `tests/test_fetch_history.py` (신규 생성)
+
+### Step 1: 실패 테스트 작성
+
+`tests/test_fetch_history.py` 파일을 새로 만든다.
+
+```python
+"""fetch_history.py의 upsert_parquet() 함수 테스트."""
+import pandas as pd
+import numpy as np
+import pytest
+from pathlib import Path
+
+
+def _make_parquet(tmp_path: Path, rows: dict) -> Path:
+    """테스트용 parquet 파일 생성 헬퍼."""
+    df = pd.DataFrame(rows)
+    df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
+    df = df.set_index("timestamp")
+    path = tmp_path / "test.parquet"
+    df.to_parquet(path)
+    return path
+
+
+def test_upsert_fills_zero_oi_with_real_value(tmp_path):
+    """기존 행의 oi_change=0.0이 신규 데이터의 실제 값으로 덮어써진다."""
+    from scripts.fetch_history import upsert_parquet
+
+    existing_path = _make_parquet(tmp_path, {
+        "timestamp": ["2026-01-01 00:00", "2026-01-01 00:15"],
+        "close": [1.0, 1.1],
+        "oi_change": [0.0, 0.0],
+        "funding_rate": [0.0, 0.0],
+    })
+
+    new_df = pd.DataFrame({
+        "close": [1.0, 1.1],
+        "oi_change": [0.05, 0.03],
+        "funding_rate": [0.0001, 0.0001],
+    }, index=pd.to_datetime(["2026-01-01 00:00", "2026-01-01 00:15"], utc=True))
+    new_df.index.name = "timestamp"
+
+    result = upsert_parquet(existing_path, new_df)
+
+    assert result.loc["2026-01-01 00:00+00:00", "oi_change"] == pytest.approx(0.05)
+    assert result.loc["2026-01-01 00:15+00:00", "oi_change"] == pytest.approx(0.03)
+
+
+def test_upsert_appends_new_rows(tmp_path):
+    """신규 타임스탬프 행이 기존 데이터 아래에 추가된다."""
+    from scripts.fetch_history import upsert_parquet
+
+    existing_path = _make_parquet(tmp_path, {
+        "timestamp": ["2026-01-01 00:00"],
+        "close": [1.0],
+        "oi_change": [0.05],
+        "funding_rate": [0.0001],
+    })
+
+    new_df = pd.DataFrame({
+        "close": [1.1],
+        "oi_change": [0.03],
+        "funding_rate": [0.0002],
+    }, index=pd.to_datetime(["2026-01-01 00:15"], utc=True))
+    new_df.index.name = "timestamp"
+
+    result = upsert_parquet(existing_path, new_df)
+
+    assert len(result) == 2
+    assert "2026-01-01 00:15+00:00" in result.index.astype(str).tolist() or \
+           pd.Timestamp("2026-01-01 00:15", tz="UTC") in result.index
+
+
+def test_upsert_keeps_nonzero_existing_oi(tmp_path):
+    """기존 행의 oi_change가 이미 0이 아니면 덮어쓰지 않는다."""
+    from scripts.fetch_history import upsert_parquet
+
+    existing_path = _make_parquet(tmp_path, {
+        "timestamp": ["2026-01-01 00:00"],
+        "close": [1.0],
+        "oi_change": [0.07],   # 이미 실제 값 존재
+        "funding_rate": [0.0003],
+    })
+
+    new_df = pd.DataFrame({
+        "close": [1.0],
+        "oi_change": [0.05],   # 다른 값으로 덮어쓰려 해도
+        "funding_rate": [0.0001],
+    }, index=pd.to_datetime(["2026-01-01 00:00"], utc=True))
+    new_df.index.name = "timestamp"
+
+    result = upsert_parquet(existing_path, new_df)
+
+    # 기존 값(0.07)이 유지되어야 한다
+    assert result.iloc[0]["oi_change"] == pytest.approx(0.07)
+
+
+def test_upsert_no_existing_file_returns_new_df(tmp_path):
+    """기존 parquet 파일이 없으면 신규 데이터를 그대로 반환한다."""
+    from scripts.fetch_history import upsert_parquet
+
+    nonexistent_path = tmp_path / "nonexistent.parquet"
+    new_df = pd.DataFrame({
+        "close": [1.0, 1.1],
+        "oi_change": [0.05, 0.03],
+        "funding_rate": [0.0001, 0.0001],
+    }, index=pd.to_datetime(["2026-01-01 00:00", "2026-01-01 00:15"], utc=True))
+    new_df.index.name = "timestamp"
+
+    result = upsert_parquet(nonexistent_path, new_df)
+
+    assert len(result) == 2
+    assert result.iloc[0]["oi_change"] == pytest.approx(0.05)
+
+
+def test_upsert_result_is_sorted_by_timestamp(tmp_path):
+    """결과 DataFrame이 timestamp 기준 오름차순 정렬되어 있다."""
+    from scripts.fetch_history import upsert_parquet
+
+    existing_path = _make_parquet(tmp_path, {
+        "timestamp": ["2026-01-01 00:15"],
+        "close": [1.1],
+        "oi_change": [0.0],
+        "funding_rate": [0.0],
+    })
+
+    new_df = pd.DataFrame({
+        "close": [1.0, 1.1, 1.2],
+        "oi_change": [0.05, 0.03, 0.02],
+        "funding_rate": [0.0001, 0.0001, 0.0002],
+    }, index=pd.to_datetime(
+        ["2026-01-01 00:00", "2026-01-01 00:15", "2026-01-01 00:30"], utc=True
+    ))
+    new_df.index.name = "timestamp"
+
+    result = upsert_parquet(existing_path, new_df)
+
+    assert result.index.is_monotonic_increasing
+    assert len(result) == 3
+```
+
+### Step 2: 테스트 실패 확인
+
+```bash
+.venv/bin/pytest tests/test_fetch_history.py -v
+```
+
+Expected: `FAILED` — `ImportError: cannot import name 'upsert_parquet' from 'scripts.fetch_history'`
+
+### Step 3: fetch_history.py에 upsert_parquet() 함수 구현
+
+`scripts/fetch_history.py`의 `main()` 함수 바로 위에 추가한다.
+
+```python
+def upsert_parquet(path: Path | str, new_df: pd.DataFrame) -> pd.DataFrame:
+    """
+    기존 parquet 파일에 신규 데이터를 Upsert(병합)한다.
+
+    규칙:
+    - 기존 행의 oi_change / funding_rate가 0.0이면 신규 값으로 덮어씀
+    - 기존 행의 oi_change / funding_rate가 이미 0이 아니면 유지
+    - 신규 타임스탬프 행은 그냥 추가
+    - 결과는 timestamp 기준 오름차순 정렬, 중복 제거
+
+    Args:
+        path: 기존 parquet 경로 (없으면 new_df 그대로 반환)
+        new_df: 새로 수집한 DataFrame (timestamp index)
+
+    Returns:
+        병합된 DataFrame
+    """
+    path = Path(path)
+    if not path.exists():
+        return new_df.sort_index()
+
+    existing = pd.read_parquet(path)
+
+    # timestamp index 통일 (tz-aware UTC)
+    if existing.index.tz is None:
+        existing.index = existing.index.tz_localize("UTC")
+    if new_df.index.tz is None:
+        new_df.index = new_df.index.tz_localize("UTC")
+
+    # 기존 데이터에서 oi_change / funding_rate가 0.0인 행만 신규 값으로 업데이트
+    UPSERT_COLS = ["oi_change", "funding_rate"]
+    overlap_idx = existing.index.intersection(new_df.index)
+
+    for col in UPSERT_COLS:
+        if col not in existing.columns or col not in new_df.columns:
+            continue
+        # 겹치는 행 중 기존 값이 0.0인 경우에만 신규 값으로 교체
+        zero_mask = existing.loc[overlap_idx, col] == 0.0
+        update_idx = overlap_idx[zero_mask]
+        if len(update_idx) > 0:
+            existing.loc[update_idx, col] = new_df.loc[update_idx, col]
+
+    # 신규 타임스탬프 행 추가 (기존에 없는 것만)
+    new_only_idx = new_df.index.difference(existing.index)
+    if len(new_only_idx) > 0:
+        existing = pd.concat([existing, new_df.loc[new_only_idx]])
+
+    return existing.sort_index()
+```
+
+### Step 4: main()에 --upsert 플래그 추가 및 저장 로직 수정
+
+`main()` 함수의 `parser` 정의 부분에 인자 추가:
+
+```python
+parser.add_argument(
+    "--no-upsert", action="store_true",
+    help="기존 parquet을 Upsert하지 않고 새로 덮어씀 (기본: Upsert 활성화)",
+)
+```
+
+그리고 단일 심볼 저장 부분:
+```python
+# 기존:
+df.to_parquet(args.output)
+
+# 변경:
+if not args.no_upsert:
+    df = upsert_parquet(args.output, df)
+df.to_parquet(args.output)
+```
+
+멀티 심볼 저장 부분도 동일하게:
+```python
+# 기존:
+merged.to_parquet(output)
+
+# 변경:
+if not args.no_upsert:
+    merged = upsert_parquet(output, merged)
+merged.to_parquet(output)
+```
+
+### Step 5: 테스트 통과 확인
+
+```bash
+.venv/bin/pytest tests/test_fetch_history.py -v
+```
+
+Expected: 전체 PASS
+
+### Step 6: 커밋
+
+```bash
+git add scripts/fetch_history.py tests/test_fetch_history.py
+git commit -m "feat: add upsert_parquet to accumulate OI/funding data incrementally"
+```
+
+---
+
+## Task 2: train_and_deploy.sh — 데이터 수집 일수 35일로 조정
+
+**Files:**
+- Modify: `scripts/train_and_deploy.sh`
+
+### Step 1: 현재 상태 확인
+
+`scripts/train_and_deploy.sh`에서 `--days 365` 부분을 찾는다.
+
+### Step 2: 수정
+
+`train_and_deploy.sh`에서 `fetch_history.py` 호출 부분을 수정한다.
+
+기존:
+```bash
+python scripts/fetch_history.py \
+    --symbols XRPUSDT BTCUSDT ETHUSDT \
+    --interval 15m \
+    --days 365 \
+    --output data/combined_15m.parquet
+```
+
+변경:
+```bash
+# OI/펀딩비 API 제한(30일) + 버퍼 5일 = 35일치 신규 수집 후 기존 parquet에 Upsert
+python scripts/fetch_history.py \
+    --symbols XRPUSDT BTCUSDT ETHUSDT \
+    --interval 15m \
+    --days 35 \
+    --output data/combined_15m.parquet
+```
+
+**이유**: 매일 실행 시 35일치만 새로 가져와 기존 누적 parquet에 Upsert한다.
+- 최초 실행 시(`data/combined_15m.parquet` 없음): 35일치로 시작
+- 이후 매일: 35일치 신규 데이터로 기존 파일의 0.0 구간을 채우고 최신 행 추가
+- 시간이 지날수록 OI/펀딩비 실제 값이 있는 구간이 1달 → 2달 → ... 로 늘어남
+
+**주의**: 최초 실행 시 캔들 데이터도 35일치만 있으므로, 첫 실행은 수동으로
+`--days 365 --no-upsert`로 전체 캔들을 먼저 수집하는 것을 권장한다.
+README에 이 내용을 추가한다.
+
+### Step 3: 커밋
+
+```bash
+git add scripts/train_and_deploy.sh
+git commit -m "feat: fetch 35 days for daily upsert instead of overwriting 365 days"
+```
+
+---
+
+## Task 3: 전체 테스트 통과 확인 및 README 업데이트
+
+### Step 1: 전체 테스트 실행
+
+```bash
+.venv/bin/pytest tests/ --ignore=tests/test_mlx_filter.py --ignore=tests/test_database.py -v
+```
+
+Expected: 전체 PASS
+
+### Step 2: README.md 업데이트
+
+**"ML 모델 학습" 섹션의 "전체 파이프라인 (권장)" 부분 아래에 아래 내용을 추가한다:**
+
+```markdown
+### 최초 실행 (캔들 전체 수집)
+
+처음 실행하거나 `data/combined_15m.parquet`가 없을 때는 전체 캔들을 먼저 수집한다.
+이후 매일 크론탭이 `train_and_deploy.sh`를 실행하면 35일치 신규 데이터가 자동으로 Upsert된다.
+
+```bash
+# 최초 1회: 1년치 캔들 전체 수집 (OI/펀딩비는 최근 30일만 실제 값, 나머지 0.0)
+python scripts/fetch_history.py \
+    --symbols XRPUSDT BTCUSDT ETHUSDT \
+    --interval 15m \
+    --days 365 \
+    --no-upsert \
+    --output data/combined_15m.parquet
+
+# 이후 매일 자동 실행 (크론탭 또는 train_and_deploy.sh):
+# 35일치 신규 데이터를 기존 파일에 Upsert → OI/펀딩비 0.0 구간이 야금야금 채워짐
+bash scripts/train_and_deploy.sh
+```
+```
+
+**"주요 기능" 섹션에 아래 항목 추가:**
+
+```markdown
+- **OI/펀딩비 누적 학습**: 매일 35일치 신규 데이터를 기존 parquet에 Upsert. 시간이 지날수록 실제 OI/펀딩비 값이 있는 학습 구간이 1달 → 2달 → 반년으로 늘어남
+```
+
+### Step 3: 최종 커밋
+
+```bash
+git add README.md
+git commit -m "docs: document OI/funding incremental accumulation strategy"
+```
+
+---
+
+## 구현 후 검증 포인트
+
+1. `data/combined_15m.parquet`에서 날짜별 `oi_change` 값 분포 확인:
+   ```python
+   import pandas as pd
+   df = pd.read_parquet("data/combined_15m.parquet")
+   print(df["oi_change"].describe())
+   print((df["oi_change"] == 0.0).sum(), "개 행이 아직 0.0")
+   ```
+2. 매일 실행 후 0.0 행 수가 줄어드는지 확인
+3. 모델 학습 시 `oi_change` / `funding_rate` 피처의 non-zero 비율이 증가하는지 확인
+
+---
+
+## 아키텍처 메모 (LXC 운영서버 관련)
+
+- **LXC 운영서버(10.1.10.24)**: 변경 없음. 모델 파일(`*.pkl` / `*.onnx`)만 받음
+- **맥미니**: `data/combined_15m.parquet`를 누적 보관. 매일 35일치 Upsert 후 학습
+- **데이터 흐름**: 맥미니 parquet 누적 → 학습 → 모델 → LXC 배포
+- **봇 실시간 OI/펀딩비**: 접근법 A(Task 1~4)에서 이미 구현됨. LXC 봇이 캔들마다 REST API로 실시간 수집

docs/plans/2026-03-02-optuna-hyperparam-tuning-design.md

@@ -0,0 +1,184 @@
+# Optuna 하이퍼파라미터 자동 튜닝 설계 문서
+
+**작성일:** 2026-03-02  
+**목표:** 봇 운영 로그/학습 결과를 바탕으로 LightGBM 하이퍼파라미터를 Optuna로 자동 탐색하고, 사람이 결과를 확인·승인한 후 재학습에 반영하는 수동 트리거 파이프라인 구축
+
+---
+
+## 배경 및 동기
+
+현재 `train_model.py`의 LightGBM 파라미터는 하드코딩되어 있다. 봇 성능이 저하되거나 데이터가 축적될 때마다 사람이 직접 파라미터를 조정해야 한다. 이를 Optuna로 자동화하되, 과적합 위험을 방지하기 위해 **사람이 결과를 먼저 확인하고 승인하는 구조**를 유지한다.
+
+---
+
+## 구현 범위 (2단계)
+
+### 1단계 (현재): LightGBM 하이퍼파라미터 튜닝
+- `scripts/tune_hyperparams.py` 신규 생성
+- Optuna + Walk-Forward AUC 목적 함수
+- 결과를 JSON + 콘솔 리포트로 출력
+
+### 2단계 (추후): 기술 지표 파라미터 확장
+- RSI 임계값, MACD 가중치, Stochastic RSI 임계값, 거래량 배수, 진입 점수 임계값 등을 탐색 공간에 추가
+- `dataset_builder.py`의 `_calc_signals()` 파라미터화 필요
+
+---
+
+## 아키텍처
+
+```
+scripts/tune_hyperparams.py
+├── load_dataset()              ← 데이터 로드 + 벡터화 데이터셋 1회 생성 (캐싱)
+├── objective(trial, dataset)   ← Optuna trial 함수
+│   ├── trial.suggest_*()       ← 하이퍼파라미터 샘플링
+│   ├── num_leaves 상한 강제    ← 2^max_depth - 1 제약
+│   └── _walk_forward_cv()      ← Walk-Forward 교차검증 → 평균 AUC 반환
+├── run_study()                 ← Optuna study 실행 (TPESampler + MedianPruner)
+├── print_report()              ← 콘솔 리포트 출력
+└── save_results()              ← JSON 저장 (models/tune_results_YYYYMMDD_HHMMSS.json)
+```
+
+---
+
+## 탐색 공간 (소규모 데이터셋 보수적 설계)
+
+| 파라미터 | 범위 | 타입 | 근거 |
+|---|---|---|---|
+| `n_estimators` | 100 ~ 600 | int | 데이터 적을 때 500+ 트리는 과적합 |
+| `learning_rate` | 0.01 ~ 0.2 | float (log) | 낮을수록 일반화 유리 |
+| `max_depth` | 2 ~ 7 | int | 트리 깊이 상한 강제 |
+| `num_leaves` | 7 ~ min(31, 2^max_depth-1) | int | **핵심**: leaf-wise 과적합 방지 |
+| `min_child_samples` | 10 ~ 50 | int | 리프당 최소 샘플 수 |
+| `subsample` | 0.5 ~ 1.0 | float | 행 샘플링 |
+| `colsample_bytree` | 0.5 ~ 1.0 | float | 열 샘플링 |
+| `reg_alpha` | 1e-4 ~ 1.0 | float (log) | L1 정규화 |
+| `reg_lambda` | 1e-4 ~ 1.0 | float (log) | L2 정규화 |
+| `time_weight_decay` | 0.5 ~ 4.0 | float | 시간 가중치 강도 |
+
+### 핵심 제약: `num_leaves <= 2^max_depth - 1`
+
+LightGBM은 leaf-wise 성장 전략을 사용하므로, `num_leaves`가 `2^max_depth - 1`을 초과하면 `max_depth` 제약이 무의미해진다. trial 내에서 `max_depth`를 먼저 샘플링한 후 `num_leaves` 상한을 동적으로 계산하여 강제한다.
+
+```python
+max_depth = trial.suggest_int("max_depth", 2, 7)
+max_leaves = min(31, 2 ** max_depth - 1)
+num_leaves = trial.suggest_int("num_leaves", 7, max_leaves)
+```
+
+---
+
+## 목적 함수: Walk-Forward AUC
+
+기존 `train_model.py`의 `walk_forward_auc()` 로직을 재활용한다. 데이터셋은 study 시작 전 1회만 생성하여 모든 trial이 공유한다 (속도 최적화).
+
+```
+전체 데이터셋 (N개 샘플)
+├── 폴드 1: 학습[0:60%] → 검증[60%:68%]
+├── 폴드 2: 학습[0:68%] → 검증[68%:76%]
+├── 폴드 3: 학습[0:76%] → 검증[76%:84%]
+├── 폴드 4: 학습[0:84%] → 검증[84%:92%]
+└── 폴드 5: 학습[0:92%] → 검증[92%:100%]
+목적 함수 = 5폴드 평균 AUC (최대화)
+```
+
+### Pruning (조기 종료)
+
+`MedianPruner` 적용: 각 폴드 완료 후 중간 AUC를 Optuna에 보고. 이전 trial들의 중앙값보다 낮으면 나머지 폴드를 건너뛰고 trial 종료. 전체 탐색 시간 ~40% 단축 효과.
+
+---
+
+## 출력 형식
+
+### 콘솔 리포트
+
+```
+============================================================
+  Optuna 튜닝 완료 | 50 trials | 소요: 28분 42초
+============================================================
+  Best AUC : 0.6234 (Trial #31)
+  Baseline : 0.5891 (현재 train_model.py 고정값)
+  개선폭   : +0.0343 (+5.8%)
+------------------------------------------------------------
+  Best Parameters:
+    n_estimators      : 320
+    learning_rate     : 0.0412
+    max_depth         : 4
+    num_leaves        : 15
+    min_child_samples : 28
+    subsample         : 0.72
+    colsample_bytree  : 0.81
+    reg_alpha         : 0.0023
+    reg_lambda         : 0.0891
+    time_weight_decay : 2.31
+------------------------------------------------------------
+  Walk-Forward 폴드별 AUC:
+    폴드 1: 0.6102
+    폴드 2: 0.6341
+    폴드 3: 0.6198
+    폴드 4: 0.6287
+    폴드 5: 0.6241
+    평균: 0.6234 ± 0.0082
+------------------------------------------------------------
+  결과 저장: models/tune_results_20260302_143022.json
+  다음 단계: python scripts/train_model.py --tuned-params models/tune_results_20260302_143022.json
+============================================================
+```
+
+### JSON 저장 (`models/tune_results_YYYYMMDD_HHMMSS.json`)
+
+```json
+{
+  "timestamp": "2026-03-02T14:30:22",
+  "n_trials": 50,
+  "elapsed_sec": 1722,
+  "baseline_auc": 0.5891,
+  "best_trial": {
+    "number": 31,
+    "auc": 0.6234,
+    "fold_aucs": [0.6102, 0.6341, 0.6198, 0.6287, 0.6241],
+    "params": { ... }
+  },
+  "all_trials": [ ... ]
+}
+```
+
+---
+
+## 사용법
+
+```bash
+# 기본 실행 (50 trials, 5폴드)
+python scripts/tune_hyperparams.py
+
+# 빠른 테스트 (10 trials, 3폴드)
+python scripts/tune_hyperparams.py --trials 10 --folds 3
+
+# 데이터 경로 지정
+python scripts/tune_hyperparams.py --data data/combined_15m.parquet --trials 100
+```
+
+---
+
+## 파일 변경 목록
+
+| 파일 | 변경 | 설명 |
+|---|---|---|
+| `scripts/tune_hyperparams.py` | **신규 생성** | Optuna 튜닝 스크립트 |
+| `requirements.txt` | **수정** | `optuna` 의존성 추가 |
+| `README.md` | **수정** | 튜닝 사용법 섹션 추가 |
+
+---
+
+## 향후 확장 (2단계)
+
+`dataset_builder.py`의 `_calc_signals()` 함수를 파라미터화하여 기술 지표 임계값도 탐색 공간에 추가:
+
+```python
+# 추가될 탐색 공간 예시
+rsi_long_threshold  = trial.suggest_int("rsi_long",  25, 40)
+rsi_short_threshold = trial.suggest_int("rsi_short", 60, 75)
+vol_surge_mult      = trial.suggest_float("vol_surge_mult", 1.2, 2.5)
+entry_threshold     = trial.suggest_int("entry_threshold", 3, 5)
+stoch_low           = trial.suggest_int("stoch_low",  10, 30)
+stoch_high          = trial.suggest_int("stoch_high", 70, 90)
+```

docs/plans/2026-03-02-optuna-hyperparam-tuning-plan.md

@@ -0,0 +1,569 @@
+# Optuna 하이퍼파라미터 자동 튜닝 Implementation Plan
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** `scripts/tune_hyperparams.py`를 신규 생성하여 Optuna + Walk-Forward AUC 기반 LightGBM 하이퍼파라미터 자동 탐색 파이프라인을 구축한다.
+
+**Architecture:** 데이터셋을 study 시작 전 1회만 생성해 캐싱하고, 각 Optuna trial에서 LightGBM 파라미터를 샘플링 → Walk-Forward 5폴드 AUC를 목적 함수로 최대화한다. `num_leaves <= 2^max_depth - 1` 제약을 코드 레벨에서 강제하여 소규모 데이터셋 과적합을 방지한다. 결과는 콘솔 리포트 + JSON 파일로 출력한다.
+
+**Tech Stack:** Python 3.11+, optuna, lightgbm, numpy, pandas, scikit-learn (기존 의존성 재활용)
+
+**설계 문서:** `docs/plans/2026-03-02-optuna-hyperparam-tuning-design.md`
+
+---
+
+## Task 1: optuna 의존성 추가
+
+**Files:**
+- Modify: `requirements.txt`
+
+**Step 1: requirements.txt에 optuna 추가**
+
+```
+optuna>=3.6.0
+```
+
+`requirements.txt` 파일 끝에 추가한다.
+
+**Step 2: 설치 확인 (로컬)**
+
+```bash
+pip install optuna
+python -c "import optuna; print(optuna.__version__)"
+```
+
+Expected: 버전 번호 출력 (예: `3.6.0`)
+
+**Step 3: Commit**
+
+```bash
+git add requirements.txt
+git commit -m "feat: add optuna dependency for hyperparameter tuning"
+```
+
+---
+
+## Task 2: `scripts/tune_hyperparams.py` 핵심 구조 생성
+
+**Files:**
+- Create: `scripts/tune_hyperparams.py`
+
+**Step 1: 파일 생성 — 전체 코드**
+
+아래 코드를 `scripts/tune_hyperparams.py`로 저장한다.
+
+```python
+"""
+Optuna를 사용한 LightGBM 하이퍼파라미터 자동 탐색.
+
+사용법:
+    python scripts/tune_hyperparams.py                          # 기본 (50 trials, 5폴드)
+    python scripts/tune_hyperparams.py --trials 10 --folds 3   # 빠른 테스트
+    python scripts/tune_hyperparams.py --data data/combined_15m.parquet --trials 100
+
+결과:
+    - 콘솔: Best Params + Walk-Forward 리포트
+    - JSON: models/tune_results_YYYYMMDD_HHMMSS.json
+"""
+import sys
+import warnings
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+import argparse
+import json
+import time
+from datetime import datetime
+
+import numpy as np
+import pandas as pd
+import lightgbm as lgb
+import optuna
+from optuna.samplers import TPESampler
+from optuna.pruners import MedianPruner
+from sklearn.metrics import roc_auc_score
+
+from src.ml_features import FEATURE_COLS
+from src.dataset_builder import generate_dataset_vectorized
+
+
+# ──────────────────────────────────────────────
+# 데이터 로드 및 데이터셋 생성 (1회 캐싱)
+# ──────────────────────────────────────────────
+
+def load_dataset(data_path: str) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
+    """
+    parquet 로드 → 벡터화 데이터셋 생성 → (X, y, w) numpy 배열 반환.
+    study 시작 전 1회만 호출하여 모든 trial이 공유한다.
+    """
+    print(f"데이터 로드: {data_path}")
+    df_raw = pd.read_parquet(data_path)
+    print(f"캔들 수: {len(df_raw):,}, 컬럼: {list(df_raw.columns)}")
+
+    base_cols = ["open", "high", "low", "close", "volume"]
+    btc_df = eth_df = None
+
+    if "close_btc" in df_raw.columns:
+        btc_df = df_raw[[c + "_btc" for c in base_cols]].copy()
+        btc_df.columns = base_cols
+        print("BTC 피처 활성화")
+
+    if "close_eth" in df_raw.columns:
+        eth_df = df_raw[[c + "_eth" for c in base_cols]].copy()
+        eth_df.columns = base_cols
+        print("ETH 피처 활성화")
+
+    df = df_raw[base_cols].copy()
+
+    print("\n데이터셋 생성 중 (1회만 실행)...")
+    dataset = generate_dataset_vectorized(df, btc_df=btc_df, eth_df=eth_df, time_weight_decay=0.0)
+
+    if dataset.empty or "label" not in dataset.columns:
+        raise ValueError("데이터셋 생성 실패: 샘플 0개")
+
+    actual_feature_cols = [c for c in FEATURE_COLS if c in dataset.columns]
+    X = dataset[actual_feature_cols].values.astype(np.float32)
+    y = dataset["label"].values.astype(np.int8)
+    w = dataset["sample_weight"].values.astype(np.float32)
+
+    pos = y.sum()
+    neg = (y == 0).sum()
+    print(f"데이터셋 완성: {len(dataset):,}개 샘플 (양성={pos:.0f}, 음성={neg:.0f})")
+    print(f"사용 피처: {len(actual_feature_cols)}개\n")
+
+    return X, y, w
+
+
+# ──────────────────────────────────────────────
+# Walk-Forward 교차검증
+# ──────────────────────────────────────────────
+
+def _walk_forward_cv(
+    X: np.ndarray,
+    y: np.ndarray,
+    w: np.ndarray,
+    params: dict,
+    n_splits: int,
+    train_ratio: float,
+    trial: optuna.Trial | None = None,
+) -> tuple[float, list[float]]:
+    """
+    Walk-Forward 교차검증으로 평균 AUC를 반환한다.
+    trial이 제공되면 각 폴드 후 Optuna에 중간 값을 보고하여 Pruning을 활성화한다.
+    """
+    n = len(X)
+    step = max(1, int(n * (1 - train_ratio) / n_splits))
+    train_end_start = int(n * train_ratio)
+
+    fold_aucs = []
+
+    for fold_idx in range(n_splits):
+        tr_end = train_end_start + fold_idx * step
+        val_end = tr_end + step
+        if val_end > n:
+            break
+
+        X_tr, y_tr, w_tr = X[:tr_end], y[:tr_end], w[:tr_end]
+        X_val, y_val = X[tr_end:val_end], y[tr_end:val_end]
+
+        # 클래스 불균형 처리: 언더샘플링 (시간 순서 유지)
+        pos_idx = np.where(y_tr == 1)[0]
+        neg_idx = np.where(y_tr == 0)[0]
+        if len(neg_idx) > len(pos_idx) and len(pos_idx) > 0:
+            rng = np.random.default_rng(42)
+            neg_idx = rng.choice(neg_idx, size=len(pos_idx), replace=False)
+        bal_idx = np.sort(np.concatenate([pos_idx, neg_idx]))
+
+        if len(bal_idx) < 20 or len(np.unique(y_val)) < 2:
+            fold_aucs.append(0.5)
+            continue
+
+        model = lgb.LGBMClassifier(**params, random_state=42, verbose=-1)
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            model.fit(X_tr[bal_idx], y_tr[bal_idx], sample_weight=w_tr[bal_idx])
+
+        proba = model.predict_proba(X_val)[:, 1]
+        auc = roc_auc_score(y_val, proba) if len(np.unique(y_val)) > 1 else 0.5
+        fold_aucs.append(auc)
+
+        # Optuna Pruning: 중간 값 보고
+        if trial is not None:
+            trial.report(float(np.mean(fold_aucs)), step=fold_idx)
+            if trial.should_prune():
+                raise optuna.TrialPruned()
+
+    mean_auc = float(np.mean(fold_aucs)) if fold_aucs else 0.5
+    return mean_auc, fold_aucs
+
+
+# ──────────────────────────────────────────────
+# Optuna 목적 함수
+# ──────────────────────────────────────────────
+
+def make_objective(
+    X: np.ndarray,
+    y: np.ndarray,
+    w: np.ndarray,
+    n_splits: int,
+    train_ratio: float,
+):
+    """클로저로 데이터셋을 캡처한 목적 함수를 반환한다."""
+
+    def objective(trial: optuna.Trial) -> float:
+        # ── 하이퍼파라미터 샘플링 ──
+        n_estimators = trial.suggest_int("n_estimators", 100, 600)
+        learning_rate = trial.suggest_float("learning_rate", 0.01, 0.2, log=True)
+        max_depth = trial.suggest_int("max_depth", 2, 7)
+
+        # 핵심 제약: num_leaves <= 2^max_depth - 1 (leaf-wise 과적합 방지)
+        max_leaves_upper = min(31, 2 ** max_depth - 1)
+        num_leaves = trial.suggest_int("num_leaves", 7, max(7, max_leaves_upper))
+
+        min_child_samples = trial.suggest_int("min_child_samples", 10, 50)
+        subsample = trial.suggest_float("subsample", 0.5, 1.0)
+        colsample_bytree = trial.suggest_float("colsample_bytree", 0.5, 1.0)
+        reg_alpha = trial.suggest_float("reg_alpha", 1e-4, 1.0, log=True)
+        reg_lambda = trial.suggest_float("reg_lambda", 1e-4, 1.0, log=True)
+
+        # time_weight_decay는 데이터셋 생성 시 적용되어야 하지만,
+        # 데이터셋을 1회 캐싱하는 구조이므로 LightGBM sample_weight 스케일로 근사한다.
+        # 실제 decay 효과는 w 배열에 이미 반영되어 있으므로 스케일 파라미터로 활용한다.
+        weight_scale = trial.suggest_float("weight_scale", 0.5, 2.0)
+        w_scaled = (w * weight_scale).astype(np.float32)
+
+        params = {
+            "n_estimators": n_estimators,
+            "learning_rate": learning_rate,
+            "max_depth": max_depth,
+            "num_leaves": num_leaves,
+            "min_child_samples": min_child_samples,
+            "subsample": subsample,
+            "colsample_bytree": colsample_bytree,
+            "reg_alpha": reg_alpha,
+            "reg_lambda": reg_lambda,
+        }
+
+        mean_auc, fold_aucs = _walk_forward_cv(
+            X, y, w_scaled, params,
+            n_splits=n_splits,
+            train_ratio=train_ratio,
+            trial=trial,
+        )
+
+        # 폴드별 AUC를 user_attrs에 저장 (결과 리포트용)
+        trial.set_user_attr("fold_aucs", fold_aucs)
+
+        return mean_auc
+
+    return objective
+
+
+# ──────────────────────────────────────────────
+# 베이스라인 AUC 측정 (현재 고정 파라미터)
+# ──────────────────────────────────────────────
+
+def measure_baseline(
+    X: np.ndarray,
+    y: np.ndarray,
+    w: np.ndarray,
+    n_splits: int,
+    train_ratio: float,
+) -> tuple[float, list[float]]:
+    """train_model.py의 현재 고정 파라미터로 베이스라인 AUC를 측정한다."""
+    baseline_params = {
+        "n_estimators": 500,
+        "learning_rate": 0.05,
+        "num_leaves": 31,
+        "min_child_samples": 15,
+        "subsample": 0.8,
+        "colsample_bytree": 0.8,
+        "reg_alpha": 0.05,
+        "reg_lambda": 0.1,
+        "max_depth": -1,  # 현재 train_model.py는 max_depth 미설정
+    }
+    print("베이스라인 측정 중 (현재 train_model.py 고정 파라미터)...")
+    return _walk_forward_cv(X, y, w, baseline_params, n_splits=n_splits, train_ratio=train_ratio)
+
+
+# ──────────────────────────────────────────────
+# 결과 출력 및 저장
+# ──────────────────────────────────────────────
+
+def print_report(
+    study: optuna.Study,
+    baseline_auc: float,
+    baseline_folds: list[float],
+    elapsed_sec: float,
+    output_path: Path,
+) -> None:
+    """콘솔에 최종 리포트를 출력한다."""
+    best = study.best_trial
+    best_auc = best.value
+    best_folds = best.user_attrs.get("fold_aucs", [])
+    improvement = best_auc - baseline_auc
+    improvement_pct = (improvement / baseline_auc * 100) if baseline_auc > 0 else 0.0
+
+    elapsed_min = int(elapsed_sec // 60)
+    elapsed_s = int(elapsed_sec % 60)
+
+    sep = "=" * 62
+    dash = "-" * 62
+
+    print(f"\n{sep}")
+    print(f"  Optuna 튜닝 완료 | {len(study.trials)} trials | 소요: {elapsed_min}분 {elapsed_s}초")
+    print(sep)
+    print(f"  Best AUC  : {best_auc:.4f}  (Trial #{best.number})")
+    print(f"  Baseline  : {baseline_auc:.4f}  (현재 train_model.py 고정값)")
+    sign = "+" if improvement >= 0 else ""
+    print(f"  개선폭    : {sign}{improvement:.4f} ({sign}{improvement_pct:.1f}%)")
+    print(dash)
+    print("  Best Parameters:")
+    for k, v in best.params.items():
+        if isinstance(v, float):
+            print(f"    {k:<22}: {v:.6f}")
+        else:
+            print(f"    {k:<22}: {v}")
+    print(dash)
+    print("  Walk-Forward 폴드별 AUC (Best Trial):")
+    for i, auc in enumerate(best_folds, 1):
+        print(f"    폴드 {i}: {auc:.4f}")
+    if best_folds:
+        print(f"    평균: {np.mean(best_folds):.4f} ± {np.std(best_folds):.4f}")
+    print(dash)
+    print("  Baseline 폴드별 AUC:")
+    for i, auc in enumerate(baseline_folds, 1):
+        print(f"    폴드 {i}: {auc:.4f}")
+    if baseline_folds:
+        print(f"    평균: {np.mean(baseline_folds):.4f} ± {np.std(baseline_folds):.4f}")
+    print(dash)
+    print(f"  결과 저장: {output_path}")
+    print(f"  다음 단계: python scripts/train_model.py --tuned-params {output_path}")
+    print(sep)
+
+
+def save_results(
+    study: optuna.Study,
+    baseline_auc: float,
+    baseline_folds: list[float],
+    elapsed_sec: float,
+    data_path: str,
+) -> Path:
+    """결과를 JSON 파일로 저장하고 경로를 반환한다."""
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    output_path = Path(f"models/tune_results_{timestamp}.json")
+    output_path.parent.mkdir(exist_ok=True)
+
+    best = study.best_trial
+
+    all_trials = []
+    for t in study.trials:
+        if t.state == optuna.trial.TrialState.COMPLETE:
+            all_trials.append({
+                "number": t.number,
+                "auc": round(t.value, 6),
+                "fold_aucs": [round(a, 6) for a in t.user_attrs.get("fold_aucs", [])],
+                "params": {k: (round(v, 6) if isinstance(v, float) else v) for k, v in t.params.items()},
+            })
+
+    result = {
+        "timestamp": datetime.now().isoformat(),
+        "data_path": data_path,
+        "n_trials_total": len(study.trials),
+        "n_trials_complete": len(all_trials),
+        "elapsed_sec": round(elapsed_sec, 1),
+        "baseline": {
+            "auc": round(baseline_auc, 6),
+            "fold_aucs": [round(a, 6) for a in baseline_folds],
+        },
+        "best_trial": {
+            "number": best.number,
+            "auc": round(best.value, 6),
+            "fold_aucs": [round(a, 6) for a in best.user_attrs.get("fold_aucs", [])],
+            "params": {k: (round(v, 6) if isinstance(v, float) else v) for k, v in best.params.items()},
+        },
+        "all_trials": all_trials,
+    }
+
+    with open(output_path, "w", encoding="utf-8") as f:
+        json.dump(result, f, indent=2, ensure_ascii=False)
+
+    return output_path
+
+
+# ──────────────────────────────────────────────
+# 메인
+# ──────────────────────────────────────────────
+
+def main():
+    parser = argparse.ArgumentParser(description="Optuna LightGBM 하이퍼파라미터 튜닝")
+    parser.add_argument("--data",   default="data/combined_15m.parquet", help="학습 데이터 경로")
+    parser.add_argument("--trials", type=int, default=50,  help="Optuna trial 수 (기본: 50)")
+    parser.add_argument("--folds",  type=int, default=5,   help="Walk-Forward 폴드 수 (기본: 5)")
+    parser.add_argument("--train-ratio", type=float, default=0.6, help="학습 구간 비율 (기본: 0.6)")
+    parser.add_argument("--no-baseline", action="store_true", help="베이스라인 측정 건너뜀")
+    args = parser.parse_args()
+
+    # 1. 데이터셋 로드 (1회)
+    X, y, w = load_dataset(args.data)
+
+    # 2. 베이스라인 측정
+    if args.no_baseline:
+        baseline_auc, baseline_folds = 0.0, []
+        print("베이스라인 측정 건너뜀 (--no-baseline)")
+    else:
+        baseline_auc, baseline_folds = measure_baseline(X, y, w, args.folds, args.train_ratio)
+        print(f"베이스라인 AUC: {baseline_auc:.4f} (폴드별: {[round(a,4) for a in baseline_folds]})\n")
+
+    # 3. Optuna study 실행
+    optuna.logging.set_verbosity(optuna.logging.WARNING)
+    sampler = TPESampler(seed=42)
+    pruner  = MedianPruner(n_startup_trials=5, n_warmup_steps=2)
+    study   = optuna.create_study(
+        direction="maximize",
+        sampler=sampler,
+        pruner=pruner,
+        study_name="lgbm_wf_auc",
+    )
+
+    objective = make_objective(X, y, w, n_splits=args.folds, train_ratio=args.train_ratio)
+
+    print(f"Optuna 탐색 시작: {args.trials} trials, {args.folds}폴드 Walk-Forward")
+    print("(진행 상황은 trial 완료마다 출력됩니다)\n")
+
+    start_time = time.time()
+
+    def _progress_callback(study: optuna.Study, trial: optuna.trial.FrozenTrial):
+        if trial.state == optuna.trial.TrialState.COMPLETE:
+            best_so_far = study.best_value
+            print(
+                f"  Trial #{trial.number:3d} | AUC={trial.value:.4f} "
+                f"| Best={best_so_far:.4f} "
+                f"| {trial.params.get('num_leaves', '?')}leaves "
+                f"depth={trial.params.get('max_depth', '?')}"
+            )
+        elif trial.state == optuna.trial.TrialState.PRUNED:
+            print(f"  Trial #{trial.number:3d} | PRUNED")
+
+    study.optimize(
+        objective,
+        n_trials=args.trials,
+        callbacks=[_progress_callback],
+        show_progress_bar=False,
+    )
+
+    elapsed = time.time() - start_time
+
+    # 4. 결과 저장 및 출력
+    output_path = save_results(study, baseline_auc, baseline_folds, elapsed, args.data)
+    print_report(study, baseline_auc, baseline_folds, elapsed, output_path)
+
+
+if __name__ == "__main__":
+    main()
+```
+
+**Step 2: 문법 오류 확인**
+
+```bash
+cd /path/to/cointrader
+python -c "import ast; ast.parse(open('scripts/tune_hyperparams.py').read()); print('문법 OK')"
+```
+
+Expected: `문법 OK`
+
+**Step 3: Commit**
+
+```bash
+git add scripts/tune_hyperparams.py
+git commit -m "feat: add Optuna Walk-Forward AUC hyperparameter tuning script"
+```
+
+---
+
+## Task 3: 동작 검증 (빠른 테스트)
+
+**Files:**
+- Read: `scripts/tune_hyperparams.py`
+
+**Step 1: 빠른 테스트 실행 (10 trials, 3폴드)**
+
+```bash
+python scripts/tune_hyperparams.py --trials 10 --folds 3 --no-baseline
+```
+
+Expected:
+- 오류 없이 10 trials 완료
+- `models/tune_results_YYYYMMDD_HHMMSS.json` 생성
+- 콘솔에 Best Params 출력
+
+**Step 2: JSON 결과 확인**
+
+```bash
+cat models/tune_results_*.json | python -m json.tool | head -40
+```
+
+Expected: `best_trial.auc`, `best_trial.params` 등 구조 확인
+
+**Step 3: Commit**
+
+```bash
+git add models/tune_results_*.json
+git commit -m "test: verify Optuna tuning pipeline with 10 trials"
+```
+
+---
+
+## Task 4: README.md 업데이트
+
+**Files:**
+- Modify: `README.md`
+
+**Step 1: ML 모델 학습 섹션에 튜닝 사용법 추가**
+
+`README.md`의 `## ML 모델 학습` 섹션 아래에 다음 내용을 추가한다:
+
+```markdown
+### 하이퍼파라미터 자동 튜닝 (Optuna)
+
+봇 성능이 저하되거나 데이터가 충분히 축적되었을 때 Optuna로 최적 파라미터를 탐색합니다.
+결과를 확인하고 직접 승인한 후 재학습에 반영하는 **수동 트리거** 방식입니다.
+
+```bash
+# 기본 실행 (50 trials, 5폴드 Walk-Forward, ~30분)
+python scripts/tune_hyperparams.py
+
+# 빠른 테스트 (10 trials, 3폴드, ~5분)
+python scripts/tune_hyperparams.py --trials 10 --folds 3
+
+# 결과 확인 후 승인하면 재학습
+python scripts/train_model.py
+```
+
+결과는 `models/tune_results_YYYYMMDD_HHMMSS.json`에 저장됩니다.
+Best Params와 베이스라인 대비 개선폭을 확인하고 직접 판단하세요.
+```
+
+**Step 2: Commit**
+
+```bash
+git add README.md
+git commit -m "docs: add Optuna hyperparameter tuning usage to README"
+```
+
+---
+
+## 검증 체크리스트
+
+- [ ] `python -c "import optuna"` 오류 없음
+- [ ] `python scripts/tune_hyperparams.py --trials 10 --folds 3 --no-baseline` 오류 없이 완료
+- [ ] `models/tune_results_*.json` 파일 생성 확인
+- [ ] JSON에 `best_trial.params`, `best_trial.fold_aucs` 포함 확인
+- [ ] 콘솔 리포트에 Best AUC, 폴드별 AUC, 파라미터 출력 확인
+- [ ] `num_leaves <= 2^max_depth - 1` 제약이 모든 trial에서 지켜지는지 JSON으로 확인
+
+---
+
+## 향후 확장 (2단계 — 별도 플랜)
+
+파이프라인 안정화 후 `dataset_builder.py`의 `_calc_signals()` 함수를 파라미터화하여 기술 지표 임계값(RSI, Stochastic RSI, 거래량 배수, 진입 점수 임계값)을 탐색 공간에 추가한다.

docs/plans/2026-03-02-realtime-oi-funding-features.md

@@ -0,0 +1,399 @@
+# 실시간 OI/펀딩비 피처 수집 구현 계획
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** 실시간 봇에서 캔들 마감 시 바이낸스 REST API로 현재 OI와 펀딩비를 수집해 ML 피처에 실제 값을 넣어 학습-추론 불일치(train-serve skew)를 해소한다.
+
+**Architecture:**
+- `exchange.py`에 `get_open_interest()`, `get_funding_rate()` 메서드 추가 (REST 호출)
+- `bot.py`의 `process_candle()`에서 캔들 마감 시 두 값을 조회하고 `build_features()` 호출 시 전달
+- `ml_features.py`의 `build_features()`가 `oi_change`, `funding_rate` 파라미터를 받아 실제 값으로 채우도록 수정
+
+**Tech Stack:** python-binance AsyncClient, aiohttp (이미 사용 중), pytest-asyncio
+
+---
+
+## Task 1: exchange.py — OI / 펀딩비 조회 메서드 추가
+
+**Files:**
+- Modify: `src/exchange.py`
+- Test: `tests/test_exchange.py`
+
+### Step 1: 실패 테스트 작성
+
+`tests/test_exchange.py` 파일에 아래 테스트를 추가한다.
+
+```python
+@pytest.mark.asyncio
+async def test_get_open_interest(exchange):
+    """get_open_interest()가 float을 반환하는지 확인."""
+    exchange.client.futures_open_interest = MagicMock(
+        return_value={"openInterest": "123456.789"}
+    )
+    result = await exchange.get_open_interest()
+    assert isinstance(result, float)
+    assert result == pytest.approx(123456.789)
+
+
+@pytest.mark.asyncio
+async def test_get_funding_rate(exchange):
+    """get_funding_rate()가 float을 반환하는지 확인."""
+    exchange.client.futures_mark_price = MagicMock(
+        return_value={"lastFundingRate": "0.0001"}
+    )
+    result = await exchange.get_funding_rate()
+    assert isinstance(result, float)
+    assert result == pytest.approx(0.0001)
+
+
+@pytest.mark.asyncio
+async def test_get_open_interest_error_returns_none(exchange):
+    """API 오류 시 None 반환 확인."""
+    from binance.exceptions import BinanceAPIException
+    exchange.client.futures_open_interest = MagicMock(
+        side_effect=BinanceAPIException(MagicMock(status_code=400), 400, '{"code":-1121,"msg":"Invalid symbol"}')
+    )
+    result = await exchange.get_open_interest()
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_get_funding_rate_error_returns_none(exchange):
+    """API 오류 시 None 반환 확인."""
+    from binance.exceptions import BinanceAPIException
+    exchange.client.futures_mark_price = MagicMock(
+        side_effect=BinanceAPIException(MagicMock(status_code=400), 400, '{"code":-1121,"msg":"Invalid symbol"}')
+    )
+    result = await exchange.get_funding_rate()
+    assert result is None
+```
+
+### Step 2: 테스트 실패 확인
+
+```bash
+pytest tests/test_exchange.py::test_get_open_interest tests/test_exchange.py::test_get_funding_rate -v
+```
+
+Expected: `FAILED` — `AttributeError: 'BinanceFuturesClient' object has no attribute 'get_open_interest'`
+
+### Step 3: exchange.py에 메서드 구현
+
+`src/exchange.py`의 `cancel_all_orders()` 메서드 아래에 추가한다.
+
+```python
+async def get_open_interest(self) -> float | None:
+    """현재 미결제약정(OI)을 조회한다. 오류 시 None 반환."""
+    loop = asyncio.get_event_loop()
+    try:
+        result = await loop.run_in_executor(
+            None,
+            lambda: self.client.futures_open_interest(symbol=self.config.symbol),
+        )
+        return float(result["openInterest"])
+    except Exception as e:
+        logger.warning(f"OI 조회 실패 (무시): {e}")
+        return None
+
+async def get_funding_rate(self) -> float | None:
+    """현재 펀딩비를 조회한다. 오류 시 None 반환."""
+    loop = asyncio.get_event_loop()
+    try:
+        result = await loop.run_in_executor(
+            None,
+            lambda: self.client.futures_mark_price(symbol=self.config.symbol),
+        )
+        return float(result["lastFundingRate"])
+    except Exception as e:
+        logger.warning(f"펀딩비 조회 실패 (무시): {e}")
+        return None
+```
+
+### Step 4: 테스트 통과 확인
+
+```bash
+pytest tests/test_exchange.py -v
+```
+
+Expected: 기존 테스트 포함 전체 PASS
+
+### Step 5: 커밋
+
+```bash
+git add src/exchange.py tests/test_exchange.py
+git commit -m "feat: add get_open_interest and get_funding_rate to BinanceFuturesClient"
+```
+
+---
+
+## Task 2: ml_features.py — build_features()에 oi/funding 파라미터 추가
+
+**Files:**
+- Modify: `src/ml_features.py`
+- Test: `tests/test_ml_features.py`
+
+### Step 1: 실패 테스트 작성
+
+`tests/test_ml_features.py`에 아래 테스트를 추가한다.
+
+```python
+def test_build_features_uses_provided_oi_funding(sample_df_with_indicators):
+    """oi_change, funding_rate 파라미터가 제공되면 실제 값이 피처에 반영된다."""
+    from src.ml_features import build_features
+    feat = build_features(
+        sample_df_with_indicators,
+        signal="LONG",
+        oi_change=0.05,
+        funding_rate=0.0002,
+    )
+    assert feat["oi_change"] == pytest.approx(0.05)
+    assert feat["funding_rate"] == pytest.approx(0.0002)
+
+
+def test_build_features_defaults_to_zero_when_not_provided(sample_df_with_indicators):
+    """oi_change, funding_rate 파라미터 미제공 시 0.0으로 채워진다."""
+    from src.ml_features import build_features
+    feat = build_features(sample_df_with_indicators, signal="LONG")
+    assert feat["oi_change"] == pytest.approx(0.0)
+    assert feat["funding_rate"] == pytest.approx(0.0)
+```
+
+### Step 2: 테스트 실패 확인
+
+```bash
+pytest tests/test_ml_features.py::test_build_features_uses_provided_oi_funding -v
+```
+
+Expected: `FAILED` — `TypeError: build_features() got an unexpected keyword argument 'oi_change'`
+
+### Step 3: ml_features.py 수정
+
+`build_features()` 시그니처와 마지막 부분을 수정한다.
+
+```python
+def build_features(
+    df: pd.DataFrame,
+    signal: str,
+    btc_df: pd.DataFrame | None = None,
+    eth_df: pd.DataFrame | None = None,
+    oi_change: float | None = None,
+    funding_rate: float | None = None,
+) -> pd.Series:
+```
+
+그리고 함수 끝의 `setdefault` 부분을 아래로 교체한다.
+
+```python
+    # 실시간에서 실제 값이 제공되면 사용, 없으면 0으로 채운다
+    base["oi_change"]    = float(oi_change)    if oi_change    is not None else 0.0
+    base["funding_rate"] = float(funding_rate) if funding_rate is not None else 0.0
+
+    return pd.Series(base)
+```
+
+기존 코드:
+```python
+    # 실시간에서는 OI/펀딩비를 수집하지 않으므로 0으로 채워 학습 피처(23개)와 일치시킨다
+    base.setdefault("oi_change", 0.0)
+    base.setdefault("funding_rate", 0.0)
+
+    return pd.Series(base)
+```
+
+### Step 4: 테스트 통과 확인
+
+```bash
+pytest tests/test_ml_features.py -v
+```
+
+Expected: 전체 PASS
+
+### Step 5: 커밋
+
+```bash
+git add src/ml_features.py tests/test_ml_features.py
+git commit -m "feat: build_features accepts oi_change and funding_rate params"
+```
+
+---
+
+## Task 3: bot.py — 캔들 마감 시 OI/펀딩비 조회 후 피처에 전달
+
+**Files:**
+- Modify: `src/bot.py`
+- Test: `tests/test_bot.py`
+
+### Step 1: 실패 테스트 작성
+
+`tests/test_bot.py`에 아래 테스트를 추가한다.
+
+```python
+@pytest.mark.asyncio
+async def test_process_candle_fetches_oi_and_funding(config, sample_df):
+    """process_candle()이 OI와 펀딩비를 조회하고 build_features에 전달하는지 확인."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot.exchange = AsyncMock()
+    bot.exchange.get_balance = AsyncMock(return_value=1000.0)
+    bot.exchange.get_position = AsyncMock(return_value=None)
+    bot.exchange.place_order = AsyncMock(return_value={"orderId": "1"})
+    bot.exchange.set_leverage = AsyncMock()
+    bot.exchange.get_open_interest = AsyncMock(return_value=5000000.0)
+    bot.exchange.get_funding_rate = AsyncMock(return_value=0.0001)
+
+    with patch("src.bot.build_features") as mock_build:
+        mock_build.return_value = pd.Series({col: 0.0 for col in __import__("src.ml_features", fromlist=["FEATURE_COLS"]).FEATURE_COLS})
+        # ML 필터는 비활성화
+        bot.ml_filter.is_model_loaded = MagicMock(return_value=False)
+        await bot.process_candle(sample_df)
+
+    # build_features가 oi_change, funding_rate 키워드 인자와 함께 호출됐는지 확인
+    assert mock_build.called
+    call_kwargs = mock_build.call_args.kwargs
+    assert "oi_change" in call_kwargs
+    assert "funding_rate" in call_kwargs
+```
+
+### Step 2: 테스트 실패 확인
+
+```bash
+pytest tests/test_bot.py::test_process_candle_fetches_oi_and_funding -v
+```
+
+Expected: `FAILED` — `AssertionError: assert 'oi_change' in {}`
+
+### Step 3: bot.py 수정
+
+`process_candle()` 메서드에서 OI/펀딩비를 조회하고 `build_features()`에 전달한다.
+
+`process_candle()` 메서드 시작 부분에 OI/펀딩비 조회를 추가한다:
+
+```python
+async def process_candle(self, df, btc_df=None, eth_df=None):
+    self.ml_filter.check_and_reload()
+
+    if not self.risk.is_trading_allowed():
+        logger.warning("리스크 한도 초과 - 거래 중단")
+        return
+
+    # 캔들 마감 시 OI/펀딩비 실시간 조회 (실패해도 0으로 폴백)
+    oi_change, funding_rate = await self._fetch_market_microstructure()
+
+    ind = Indicators(df)
+    df_with_indicators = ind.calculate_all()
+    raw_signal = ind.get_signal(df_with_indicators)
+    # ... (이하 동일)
+```
+
+그리고 `build_features()` 호출 부분 두 곳을 모두 수정한다:
+
+```python
+features = build_features(
+    df_with_indicators, signal,
+    btc_df=btc_df, eth_df=eth_df,
+    oi_change=oi_change, funding_rate=funding_rate,
+)
+```
+
+`_fetch_market_microstructure()` 메서드를 추가한다:
+
+```python
+async def _fetch_market_microstructure(self) -> tuple[float, float]:
+    """OI 변화율과 펀딩비를 실시간으로 조회한다. 실패 시 0.0으로 폴백."""
+    oi_val, fr_val = await asyncio.gather(
+        self.exchange.get_open_interest(),
+        self.exchange.get_funding_rate(),
+        return_exceptions=True,
+    )
+    oi_float = float(oi_val) if isinstance(oi_val, (int, float)) else 0.0
+    fr_float = float(fr_val) if isinstance(fr_val, (int, float)) else 0.0
+
+    # OI는 절대값이므로 이전 값 대비 변화율로 변환
+    oi_change = self._calc_oi_change(oi_float)
+    logger.debug(f"OI={oi_float:.0f}, OI변화율={oi_change:.6f}, 펀딩비={fr_float:.6f}")
+    return oi_change, fr_float
+```
+
+`_calc_oi_change()` 메서드와 `_prev_oi` 상태를 추가한다:
+
+`__init__()` 에 추가:
+```python
+self._prev_oi: float | None = None  # OI 변화율 계산용 이전 값
+```
+
+메서드 추가:
+```python
+def _calc_oi_change(self, current_oi: float) -> float:
+    """이전 OI 대비 변화율을 계산한다. 첫 캔들은 0.0 반환."""
+    if self._prev_oi is None or self._prev_oi == 0.0:
+        self._prev_oi = current_oi
+        return 0.0
+    change = (current_oi - self._prev_oi) / self._prev_oi
+    self._prev_oi = current_oi
+    return change
+```
+
+### Step 4: 테스트 통과 확인
+
+```bash
+pytest tests/test_bot.py -v
+```
+
+Expected: 전체 PASS
+
+### Step 5: 커밋
+
+```bash
+git add src/bot.py tests/test_bot.py
+git commit -m "feat: fetch realtime OI and funding rate on candle close for ML features"
+```
+
+---
+
+## Task 4: 전체 테스트 통과 확인 및 README 업데이트
+
+### Step 1: 전체 테스트 실행
+
+```bash
+bash scripts/run_tests.sh
+```
+
+Expected: 전체 PASS (새 테스트 포함)
+
+### Step 2: README.md 업데이트
+
+`README.md`의 "주요 기능" 섹션에서 ML 피처 설명을 수정한다.
+
+기존:
+```
+- **23개 ML 피처**: XRP 기술 지표 13개 + BTC/ETH 수익률·상대강도 8개 + OI 변화율·펀딩비 2개 (실시간 미수집 항목은 0으로 채움)
+```
+
+변경:
+```
+- **23개 ML 피처**: XRP 기술 지표 13개 + BTC/ETH 수익률·상대강도 8개 + OI 변화율·펀딩비 2개 (캔들 마감 시 REST API로 실시간 수집)
+```
+
+### Step 3: 최종 커밋
+
+```bash
+git add README.md
+git commit -m "docs: update README to reflect realtime OI/funding rate collection"
+```
+
+---
+
+## 구현 후 검증 포인트
+
+1. 봇 실행 로그에서 `OI=xxx, OI변화율=xxx, 펀딩비=xxx` 라인이 15분마다 출력되는지 확인
+2. API 오류(네트워크 단절 등) 시 `WARNING: OI 조회 실패 (무시)` 로그 후 0.0으로 폴백해 봇이 정상 동작하는지 확인
+3. `build_features()` 호출 시 `oi_change`, `funding_rate`가 실제 값으로 채워지는지 로그 확인
+
+---
+
+## 다음 단계: 접근법 B (OI/펀딩비 누적 저장)
+
+A 완료 후 진행할 계획:
+- `scripts/fetch_history.py` 실행 시 기존 parquet에 새 30일치를 **append(중복 제거)** 방식으로 저장
+- 시간이 지날수록 OI/펀딩비 학습 데이터가 누적되어 모델 품질 향상
+- 별도 플랜 문서로 작성 예정

docs/plans/2026-03-02-reverse-signal-reenter-design.md

@@ -0,0 +1,125 @@
+# 반대 시그널 시 청산 후 즉시 재진입 설계
+
+- **날짜**: 2026-03-02
+- **파일**: `src/bot.py`
+- **상태**: 설계 완료, 구현 대기
+
+---
+
+## 배경
+
+현재 `TradingBot.process_candle`은 반대 방향 시그널이 오면 기존 포지션을 청산만 하고 종료한다.
+새 포지션은 다음 캔들에서 시그널이 다시 나와야 잡힌다.
+
+```
+현재: 반대 시그널 → 청산 → 다음 캔들 대기
+목표: 반대 시그널 → 청산 → (ML 필터 통과 시) 즉시 반대 방향 재진입
+```
+
+같은 방향 시그널이 오거나 HOLD이면 기존 포지션을 그대로 유지한다.
+
+---
+
+## 요구사항
+
+| 항목 | 결정 |
+|------|------|
+| 포지션 크기 | 재진입 시점 잔고 + 동적 증거금 비율로 새로 계산 |
+| SL/TP | 청산 시 기존 주문 전부 취소, 재진입 시 새로 설정 |
+| ML 필터 | 재진입에도 동일하게 적용 (차단 시 청산만 하고 대기) |
+| 같은 방향 시그널 | 포지션 유지 (변경 없음) |
+| HOLD 시그널 | 포지션 유지 (변경 없음) |
+
+---
+
+## 설계
+
+### 변경 범위
+
+`src/bot.py` 한 파일만 수정한다.
+
+1. `_close_and_reenter` 메서드 신규 추가
+2. `process_candle` 내 반대 시그널 분기에서 `_close_position` 대신 `_close_and_reenter` 호출
+
+### 데이터 흐름
+
+```
+process_candle()
+  └─ 반대 시그널 감지
+       └─ _close_and_reenter(position, signal, df, btc_df, eth_df)
+            ├─ _close_position(position)          # 청산 + cancel_all_orders
+            ├─ risk.can_open_new_position() 체크
+            │    └─ 불가 → 로그 + 종료
+            ├─ ML 필터 체크 (ml_filter.is_model_loaded())
+            │    ├─ 차단 → 로그 + 종료 (포지션 없는 상태로 대기)
+            │    └─ 통과 → 계속
+            └─ _open_position(signal, df)          # 재진입 + 새 SL/TP 설정
+```
+
+### `process_candle` 수정
+
+```python
+# 변경 전
+elif position is not None:
+    pos_side = "LONG" if float(position["positionAmt"]) > 0 else "SHORT"
+    if (pos_side == "LONG" and signal == "SHORT") or \
+       (pos_side == "SHORT" and signal == "LONG"):
+        await self._close_position(position)
+
+# 변경 후
+elif position is not None:
+    pos_side = "LONG" if float(position["positionAmt"]) > 0 else "SHORT"
+    if (pos_side == "LONG" and signal == "SHORT") or \
+       (pos_side == "SHORT" and signal == "LONG"):
+        await self._close_and_reenter(position, signal, df_with_indicators, btc_df, eth_df)
+```
+
+### 신규 메서드 `_close_and_reenter`
+
+```python
+async def _close_and_reenter(
+    self,
+    position: dict,
+    signal: str,
+    df,
+    btc_df=None,
+    eth_df=None,
+) -> None:
+    """기존 포지션을 청산하고, ML 필터 통과 시 반대 방향으로 즉시 재진입한다."""
+    await self._close_position(position)
+
+    if not self.risk.can_open_new_position():
+        logger.info("최대 포지션 수 도달 — 재진입 건너뜀")
+        return
+
+    if self.ml_filter.is_model_loaded():
+        features = build_features(df, signal, btc_df=btc_df, eth_df=eth_df)
+        if not self.ml_filter.should_enter(features):
+            logger.info(f"ML 필터 차단: {signal} 재진입 무시")
+            return
+
+    await self._open_position(signal, df)
+```
+
+---
+
+## 엣지 케이스
+
+| 상황 | 처리 |
+|------|------|
+| 청산 후 ML 필터 차단 | 청산만 하고 포지션 없는 상태로 대기 |
+| 청산 후 잔고 부족 (명목금액 미달) | `_open_position` 내부 경고 후 건너뜀 (기존 로직) |
+| 청산 후 최대 포지션 수 초과 | 재진입 건너뜀 |
+| 같은 방향 시그널 | 포지션 유지 (변경 없음) |
+| HOLD 시그널 | 포지션 유지 (변경 없음) |
+| 봇 재시작 후 포지션 복구 | `_recover_position` 로직 변경 없음 |
+
+---
+
+## 영향 없는 코드
+
+- `_close_position` — 변경 없음
+- `_open_position` — 변경 없음
+- `_recover_position` — 변경 없음
+- `RiskManager` — 변경 없음
+- `MLFilter` — 변경 없음

docs/plans/2026-03-02-reverse-signal-reenter-plan.md

@@ -0,0 +1,269 @@
+# 반대 시그널 시 청산 후 즉시 재진입 구현 플랜
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** 반대 방향 시그널이 오면 기존 포지션을 청산하고 ML 필터 통과 시 즉시 반대 방향으로 재진입한다.
+
+**Architecture:** `src/bot.py`에 `_close_and_reenter` 메서드를 추가하고, `process_candle`의 반대 시그널 분기에서 이를 호출한다. 기존 `_close_position`과 `_open_position`을 그대로 재사용하므로 중복 없음.
+
+**Tech Stack:** Python 3.12, pytest, unittest.mock
+
+---
+
+## 테스트 스크립트
+
+각 태스크 단계마다 아래 스크립트로 테스트를 실행한다.
+
+```bash
+# Task 1 — 신규 테스트 실행 (구현 전, FAIL 확인용)
+bash scripts/test_reverse_reenter.sh 1
+
+# Task 2 — _close_and_reenter 메서드 테스트 (구현 후, PASS 확인)
+bash scripts/test_reverse_reenter.sh 2
+
+# Task 3 — process_candle 분기 테스트 (수정 후, PASS 확인)
+bash scripts/test_reverse_reenter.sh 3
+
+# test_bot.py 전체
+bash scripts/test_reverse_reenter.sh bot
+
+# 전체 테스트 스위트
+bash scripts/test_reverse_reenter.sh all
+```
+
+---
+
+## 참고 파일
+
+- 설계 문서: `docs/plans/2026-03-02-reverse-signal-reenter-design.md`
+- 구현 대상: `src/bot.py`
+- 기존 테스트: `tests/test_bot.py`
+
+---
+
+## Task 1: `_close_and_reenter` 테스트 작성
+
+**Files:**
+- Modify: `tests/test_bot.py`
+
+### Step 1: 테스트 3개 추가
+
+`tests/test_bot.py` 맨 아래에 다음 테스트를 추가한다.
+
+```python
+@pytest.mark.asyncio
+async def test_close_and_reenter_calls_open_when_ml_passes(config, sample_df):
+    """반대 시그널 + ML 필터 통과 시 청산 후 재진입해야 한다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot._close_position = AsyncMock()
+    bot._open_position = AsyncMock()
+    bot.ml_filter = MagicMock()
+    bot.ml_filter.is_model_loaded.return_value = True
+    bot.ml_filter.should_enter.return_value = True
+
+    position = {"positionAmt": "100", "entryPrice": "0.5", "markPrice": "0.52"}
+    await bot._close_and_reenter(position, "SHORT", sample_df)
+
+    bot._close_position.assert_awaited_once_with(position)
+    bot._open_position.assert_awaited_once_with("SHORT", sample_df)
+
+
+@pytest.mark.asyncio
+async def test_close_and_reenter_skips_open_when_ml_blocks(config, sample_df):
+    """ML 필터 차단 시 청산만 하고 재진입하지 않아야 한다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot._close_position = AsyncMock()
+    bot._open_position = AsyncMock()
+    bot.ml_filter = MagicMock()
+    bot.ml_filter.is_model_loaded.return_value = True
+    bot.ml_filter.should_enter.return_value = False
+
+    position = {"positionAmt": "100", "entryPrice": "0.5", "markPrice": "0.52"}
+    await bot._close_and_reenter(position, "SHORT", sample_df)
+
+    bot._close_position.assert_awaited_once_with(position)
+    bot._open_position.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_close_and_reenter_skips_open_when_max_positions_reached(config, sample_df):
+    """최대 포지션 수 도달 시 청산만 하고 재진입하지 않아야 한다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot._close_position = AsyncMock()
+    bot._open_position = AsyncMock()
+    bot.risk = MagicMock()
+    bot.risk.can_open_new_position.return_value = False
+
+    position = {"positionAmt": "100", "entryPrice": "0.5", "markPrice": "0.52"}
+    await bot._close_and_reenter(position, "SHORT", sample_df)
+
+    bot._close_position.assert_awaited_once_with(position)
+    bot._open_position.assert_not_called()
+```
+
+### Step 2: 테스트 실행 — 실패 확인
+
+```bash
+bash scripts/test_reverse_reenter.sh 1
+```
+
+예상 결과: `AttributeError: 'TradingBot' object has no attribute '_close_and_reenter'` 로 3개 FAIL
+
+---
+
+## Task 2: `_close_and_reenter` 메서드 구현
+
+**Files:**
+- Modify: `src/bot.py:148` (`_close_position` 메서드 바로 아래에 추가)
+
+### Step 1: `_close_position` 다음에 메서드 추가
+
+`src/bot.py`에서 `_close_position` 메서드(148~167번째 줄) 바로 뒤에 다음을 추가한다.
+
+```python
+    async def _close_and_reenter(
+        self,
+        position: dict,
+        signal: str,
+        df,
+        btc_df=None,
+        eth_df=None,
+    ) -> None:
+        """기존 포지션을 청산하고, ML 필터 통과 시 반대 방향으로 즉시 재진입한다."""
+        await self._close_position(position)
+
+        if not self.risk.can_open_new_position():
+            logger.info("최대 포지션 수 도달 — 재진입 건너뜀")
+            return
+
+        if self.ml_filter.is_model_loaded():
+            features = build_features(df, signal, btc_df=btc_df, eth_df=eth_df)
+            if not self.ml_filter.should_enter(features):
+                logger.info(f"ML 필터 차단: {signal} 재진입 무시")
+                return
+
+        await self._open_position(signal, df)
+```
+
+### Step 2: 테스트 실행 — 통과 확인
+
+```bash
+bash scripts/test_reverse_reenter.sh 2
+```
+
+예상 결과: 3개 PASS
+
+### Step 3: 커밋
+
+```bash
+git add src/bot.py tests/test_bot.py
+git commit -m "feat: add _close_and_reenter method for reverse signal handling"
+```
+
+---
+
+## Task 3: `process_candle` 분기 수정
+
+**Files:**
+- Modify: `src/bot.py:83-85`
+
+### Step 1: 기존 분기 테스트 추가
+
+`tests/test_bot.py`에 다음 테스트를 추가한다.
+
+```python
+@pytest.mark.asyncio
+async def test_process_candle_calls_close_and_reenter_on_reverse_signal(config, sample_df):
+    """반대 시그널 시 process_candle이 _close_and_reenter를 호출해야 한다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot.exchange = AsyncMock()
+    bot.exchange.get_position = AsyncMock(return_value={
+        "positionAmt": "100",
+        "entryPrice": "0.5",
+        "markPrice": "0.52",
+    })
+    bot._close_and_reenter = AsyncMock()
+    bot.ml_filter = MagicMock()
+    bot.ml_filter.is_model_loaded.return_value = False
+    bot.ml_filter.should_enter.return_value = True
+
+    with patch("src.bot.Indicators") as MockInd:
+        mock_ind = MagicMock()
+        mock_ind.calculate_all.return_value = sample_df
+        mock_ind.get_signal.return_value = "SHORT"  # 현재 LONG 포지션에 반대 시그널
+        MockInd.return_value = mock_ind
+        await bot.process_candle(sample_df)
+
+    bot._close_and_reenter.assert_awaited_once()
+    call_args = bot._close_and_reenter.call_args
+    assert call_args.args[1] == "SHORT"
+```
+
+### Step 2: 테스트 실행 — 실패 확인
+
+```bash
+bash scripts/test_reverse_reenter.sh 3
+```
+
+예상 결과: FAIL (`_close_and_reenter`가 아직 호출되지 않음)
+
+### Step 3: `process_candle` 수정
+
+`src/bot.py`에서 아래 부분을 찾아 수정한다.
+
+```python
+# 변경 전 (81~85번째 줄 근처)
+        elif position is not None:
+            pos_side = "LONG" if float(position["positionAmt"]) > 0 else "SHORT"
+            if (pos_side == "LONG" and signal == "SHORT") or \
+               (pos_side == "SHORT" and signal == "LONG"):
+                await self._close_position(position)
+
+# 변경 후
+        elif position is not None:
+            pos_side = "LONG" if float(position["positionAmt"]) > 0 else "SHORT"
+            if (pos_side == "LONG" and signal == "SHORT") or \
+               (pos_side == "SHORT" and signal == "LONG"):
+                await self._close_and_reenter(
+                    position, signal, df_with_indicators, btc_df=btc_df, eth_df=eth_df
+                )
+```
+
+### Step 4: 전체 테스트 실행 — 통과 확인
+
+```bash
+bash scripts/test_reverse_reenter.sh bot
+```
+
+예상 결과: 전체 PASS (기존 테스트 포함)
+
+### Step 5: 커밋
+
+```bash
+git add src/bot.py tests/test_bot.py
+git commit -m "feat: call _close_and_reenter on reverse signal in process_candle"
+```
+
+---
+
+## Task 4: 전체 테스트 스위트 확인
+
+### Step 1: 전체 테스트 실행
+
+```bash
+bash scripts/test_reverse_reenter.sh all
+```
+
+예상 결과: 모든 테스트 PASS
+
+### Step 2: 실패 테스트 있으면 수정 후 재실행
+
+실패가 있으면 원인을 파악하고 수정한다. 기존 테스트를 깨뜨리지 않도록 주의.

docs/plans/2026-03-02-rs-divide-mlx-nan-fix.md

@@ -0,0 +1,203 @@
+# RS np.divide 복구 / MLX NaN-Safe 통계 저장 구현 계획
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** RS(상대강도) 계산의 epsilon 폭발 이상치를 `np.divide` 방식으로 제거하고, MLXFilter의 `self._mean`/`self._std`에 NaN이 잔류하는 근본 허점을 차단한다.
+
+**Architecture:**
+- `src/dataset_builder.py`: `xrp_btc_rs_raw` / `xrp_eth_rs_raw` 계산을 `np.divide(..., where=...)` 방식으로 교체. 분모(btc_r1, eth_r1)가 0이면 결과를 0.0으로 채워 rolling zscore 윈도우 오염을 방지한다.
+- `src/mlx_filter.py`: `fit()` 내부에서 `self._mean`/`self._std`를 저장하기 전에 `nan_to_num`을 적용해 전체-NaN 컬럼(OI 초반 구간 등)이 `predict_proba` 시점까지 NaN을 전파하지 않도록 한다.
+
+**Tech Stack:** numpy, pandas, pytest, mlx(Apple Silicon 전용 — MLX 테스트는 Mac에서만 실행)
+
+---
+
+### Task 1: `dataset_builder.py` — RS 계산을 `np.divide` 방식으로 교체
+
+**Files:**
+- Modify: `src/dataset_builder.py:245-246`
+- Test: `tests/test_dataset_builder.py`
+
+**배경:**
+`btc_r1 = 0.0`(15분 동안 BTC 가격 변동 없음)일 때 `xrp_r1 / (btc_r1 + 1e-8)`는 최대 수백만의 이상치를 만든다. 이 이상치가 288캔들 rolling zscore 윈도우에 들어가면 나머지 287개 값이 전부 0에 가깝게 압사된다.
+
+**Step 1: 기존 테스트 실행 (기준선 확인)**
+
+```bash
+python -m pytest tests/test_dataset_builder.py -v
+```
+
+Expected: 모든 테스트 PASS (변경 전 기준선)
+
+**Step 2: RS 제로-분모 테스트 작성**
+
+`tests/test_dataset_builder.py` 파일 끝에 추가:
+
+```python
+def test_rs_zero_denominator():
+    """btc_r1=0일 때 RS가 inf/nan이 아닌 0.0이어야 한다 (np.divide 방식 검증)."""
+    import numpy as np
+    import pandas as pd
+    from src.dataset_builder import _calc_features_vectorized, _calc_signals, _calc_indicators
+
+    n = 500
+    np.random.seed(7)
+    # XRP close: 약간의 변동
+    xrp_close = np.cumprod(1 + np.random.randn(n) * 0.001) * 1.0
+    xrp_df = pd.DataFrame({
+        "open":   xrp_close * 0.999,
+        "high":   xrp_close * 1.005,
+        "low":    xrp_close * 0.995,
+        "close":  xrp_close,
+        "volume": np.random.rand(n) * 1000 + 500,
+    })
+    # BTC close: 완전히 고정 → btc_r1 = 0.0
+    btc_close = np.ones(n) * 50000.0
+    btc_df = pd.DataFrame({
+        "open":   btc_close,
+        "high":   btc_close,
+        "low":    btc_close,
+        "close":  btc_close,
+        "volume": np.random.rand(n) * 1000 + 500,
+    })
+
+    from src.dataset_builder import generate_dataset_vectorized
+    result = generate_dataset_vectorized(xrp_df, btc_df=btc_df)
+
+    if result.empty:
+        pytest.skip("신호 없음")
+
+    assert "xrp_btc_rs" in result.columns, "xrp_btc_rs 컬럼이 있어야 함"
+    assert not result["xrp_btc_rs"].isin([np.inf, -np.inf]).any(), \
+        "xrp_btc_rs에 inf가 있으면 안 됨"
+    assert not result["xrp_btc_rs"].isna().all(), \
+        "xrp_btc_rs가 전부 nan이면 안 됨"
+```
+
+**Step 3: 테스트 실행 (FAIL 확인)**
+
+```bash
+python -m pytest tests/test_dataset_builder.py::test_rs_zero_denominator -v
+```
+
+Expected: FAIL — `xrp_btc_rs에 inf가 있으면 안 됨` (현재 epsilon 방식은 inf 대신 수백만 이상치를 만들어 rolling zscore 후 nan이 될 수 있음)
+
+> 참고: 현재 코드는 inf를 직접 만들지 않을 수도 있다. 하지만 rolling zscore 후 nan이 생기거나 이상치가 남아있는지 확인하는 것이 목적이다. PASS가 나오더라도 Step 4를 진행한다.
+
+**Step 4: `dataset_builder.py` 245~246줄 수정**
+
+`src/dataset_builder.py`의 아래 두 줄을:
+
+```python
+        xrp_btc_rs_raw = (xrp_r1 / (btc_r1 + 1e-8)).astype(np.float32)
+        xrp_eth_rs_raw = (xrp_r1 / (eth_r1 + 1e-8)).astype(np.float32)
+```
+
+다음으로 교체:
+
+```python
+        xrp_btc_rs_raw = np.divide(
+            xrp_r1, btc_r1,
+            out=np.zeros_like(xrp_r1),
+            where=(btc_r1 != 0),
+        ).astype(np.float32)
+        xrp_eth_rs_raw = np.divide(
+            xrp_r1, eth_r1,
+            out=np.zeros_like(xrp_r1),
+            where=(eth_r1 != 0),
+        ).astype(np.float32)
+```
+
+**Step 5: 전체 테스트 실행 (PASS 확인)**
+
+```bash
+python -m pytest tests/test_dataset_builder.py -v
+```
+
+Expected: 모든 테스트 PASS
+
+**Step 6: 커밋**
+
+```bash
+git add src/dataset_builder.py tests/test_dataset_builder.py
+git commit -m "fix: RS 계산을 np.divide(where=) 방식으로 교체 — epsilon 이상치 폭발 차단"
+```
+
+---
+
+### Task 2: `mlx_filter.py` — `self._mean`/`self._std` 저장 전 `nan_to_num` 적용
+
+**Files:**
+- Modify: `src/mlx_filter.py:145-146`
+- Test: `tests/test_mlx_filter.py` (기존 `test_fit_with_nan_features` 활용)
+
+**배경:**
+현재 코드는 `self._mean = np.nanmean(X_np, axis=0)`으로 저장한다. 전체가 NaN인 컬럼(Walk-Forward 초반 11개월의 OI 데이터)이 있으면 `np.nanmean`은 해당 컬럼의 평균으로 NaN을 반환한다. 이 NaN이 `self._mean`에 저장되면 `predict_proba` 시점에 `(X_np - self._mean)`이 NaN이 되어 OI 데이터를 영원히 활용하지 못한다.
+
+**Step 1: 기존 테스트 실행 (기준선 확인)**
+
+```bash
+python -m pytest tests/test_mlx_filter.py -v
+```
+
+Expected: 모든 테스트 PASS (MLX 없는 환경에서는 전체 SKIP)
+
+**Step 2: `mlx_filter.py` 145~146줄 수정**
+
+`src/mlx_filter.py`의 아래 두 줄을:
+
+```python
+        self._mean = np.nanmean(X_np, axis=0)
+        self._std  = np.nanstd(X_np, axis=0) + 1e-8
+```
+
+다음으로 교체:
+
+```python
+        mean_vals  = np.nanmean(X_np, axis=0)
+        self._mean = np.nan_to_num(mean_vals, nan=0.0)   # 전체-NaN 컬럼 → 평균 0.0
+        std_vals   = np.nanstd(X_np, axis=0)
+        self._std  = np.nan_to_num(std_vals, nan=1.0) + 1e-8  # 전체-NaN 컬럼 → std 1.0
+```
+
+**Step 3: 테스트 실행 (PASS 확인)**
+
+```bash
+python -m pytest tests/test_mlx_filter.py::test_fit_with_nan_features -v
+```
+
+Expected: PASS (MLX 없는 환경에서는 SKIP)
+
+**Step 4: 전체 테스트 실행**
+
+```bash
+python -m pytest tests/test_mlx_filter.py -v
+```
+
+Expected: 모든 테스트 PASS (또는 SKIP)
+
+**Step 5: 커밋**
+
+```bash
+git add src/mlx_filter.py
+git commit -m "fix: MLXFilter self._mean/std 저장 전 nan_to_num 적용 — 전체-NaN 컬럼 predict_proba 오염 차단"
+```
+
+---
+
+### Task 3: 전체 테스트 통과 확인
+
+**Step 1: 전체 테스트 실행**
+
+```bash
+python -m pytest tests/ -v --tb=short 2>&1 | tail -40
+```
+
+Expected: 모든 테스트 PASS (MLX 관련은 SKIP 허용)
+
+**Step 2: 최종 커밋 (필요 시)**
+
+```bash
+git add -A
+git commit -m "chore: RS epsilon 폭발 차단 + MLX NaN-Safe 통계 저장 통합"
+```

docs/plans/2026-03-02-user-data-stream-tp-sl-detection-design.md

@@ -0,0 +1,300 @@
+# User Data Stream TP/SL 감지 설계
+
+**날짜:** 2026-03-02  
+**목적:** Binance Futures User Data Stream을 도입하여 TP/SL 작동을 실시간 감지하고, 순수익(Net PnL)을 기록하며, Discord에 상세 청산 알림을 전송한다.
+
+---
+
+## 배경 및 문제
+
+기존 봇은 매 캔들 마감마다 `get_position()`을 폴링하여 포지션 소멸 여부를 확인하는 방식이었다. 이 구조의 한계:
+
+1. **TP/SL 작동 후 최대 15분 지연** — 캔들 마감 전까지 감지 불가
+2. **청산 원인 구분 불가** — TP인지 SL인지 수동 청산인지 알 수 없음
+3. **PnL 기록 누락** — `_close_position()`을 봇이 직접 호출하지 않으면 `record_pnl()` 미실행
+4. **Discord 알림 누락** — 동일 이유로 `notify_close()` 미호출
+
+---
+
+## 선택한 접근 방식
+
+**방식 A: `python-binance` 내장 User Data Stream + 30분 수동 keepalive 보강**
+
+- 기존 `BinanceSocketManager` 활용으로 추가 의존성 없음
+- `futures_user_socket(listenKey)`로 User Data Stream 연결
+- 별도 30분 keepalive 백그라운드 태스크로 안정성 보강
+- `while True: try-except` 무한 재연결 루프로 네트워크 단절 복구
+
+---
+
+## 전체 아키텍처
+
+### 파일 변경 목록
+
+| 파일 | 변경 유형 | 내용 |
+|------|----------|------|
+| `src/user_data_stream.py` | **신규** | User Data Stream 전담 클래스 |
+| `src/bot.py` | 수정 | `UserDataStream` 초기화, `run()` 병렬 실행, `_on_position_closed()` 콜백, `_entry_price`/`_entry_quantity` 상태 추가 |
+| `src/exchange.py` | 수정 | `create_listen_key()`, `keepalive_listen_key()`, `delete_listen_key()` 메서드 추가 |
+| `src/notifier.py` | 수정 | `notify_close()`에 `close_reason`, `estimated_pnl`, `net_pnl` 파라미터 추가 |
+| `src/risk_manager.py` | 수정 | `record_pnl()`이 net_pnl을 받도록 유지 (인터페이스 변경 없음) |
+
+### 실행 흐름
+
+```
+bot.run()
+  └── AsyncClient 단일 인스턴스 생성
+        └── asyncio.gather()
+              ├── MultiSymbolStream.start(client)   ← 기존 캔들 스트림
+              └── UserDataStream.start()             ← 신규
+                    ├── [백그라운드] _keepalive_loop()  30분마다 PUT /listenKey
+                    └── [메인루프]  while True:
+                                    try:
+                                      listenKey 발급
+                                      futures_user_socket() 연결
+                                      async for msg: _handle_message()
+                                    except CancelledError: break
+                                    except Exception: sleep(5) → 재연결
+```
+
+---
+
+## 섹션 1: UserDataStream 클래스 (`src/user_data_stream.py`)
+
+### 상수
+
+```python
+KEEPALIVE_INTERVAL = 30 * 60   # 30분 (listenKey 만료 60분의 절반)
+RECONNECT_DELAY    = 5         # 재연결 대기 초
+```
+
+### listenKey 생명주기
+
+| 단계 | API | 시점 |
+|------|-----|------|
+| 발급 | `POST /fapi/v1/listenKey` | 연결 시작 / 재연결 시 |
+| 갱신 | `PUT /fapi/v1/listenKey` | 30분마다 (백그라운드 태스크) |
+| 삭제 | `DELETE /fapi/v1/listenKey` | 봇 정상 종료 시 (`CancelledError`) |
+
+### 재연결 로직
+
+```python
+while True:
+    try:
+        listen_key = await exchange.create_listen_key()
+        keepalive_task = asyncio.create_task(_keepalive_loop(listen_key))
+        async with bm.futures_user_socket(listen_key):
+            async for msg:
+                await _handle_message(msg)
+    except asyncio.CancelledError:
+        await exchange.delete_listen_key(listen_key)
+        keepalive_task.cancel()
+        break
+    except Exception as e:
+        logger.warning(f"User Data Stream 끊김: {e}, {RECONNECT_DELAY}초 후 재연결")
+        keepalive_task.cancel()
+        await asyncio.sleep(RECONNECT_DELAY)
+        # while True 상단으로 돌아가 listenKey 재발급
+```
+
+### keepalive 백그라운드 태스크
+
+```python
+async def _keepalive_loop(listen_key: str):
+    while True:
+        await asyncio.sleep(KEEPALIVE_INTERVAL)
+        try:
+            await exchange.keepalive_listen_key(listen_key)
+            logger.debug("listenKey 갱신 완료")
+        except Exception:
+            logger.warning("listenKey 갱신 실패 → 재연결 루프가 처리")
+            break  # 재연결 루프가 새 태스크 생성
+```
+
+---
+
+## 섹션 2: 이벤트 파싱 로직
+
+### 페이로드 구조 (Binance Futures ORDER_TRADE_UPDATE)
+
+주문 상세 정보는 최상위가 아닌 **내부 `"o"` 딕셔너리에 중첩**되어 있다.
+
+```json
+{
+  "e": "ORDER_TRADE_UPDATE",
+  "o": {
+    "x": "TRADE",          // Execution Type
+    "X": "FILLED",         // Order Status
+    "o": "TAKE_PROFIT_MARKET",  // Order Type
+    "R": true,             // reduceOnly
+    "rp": "0.48210000",    // realizedProfit
+    "n": "0.02100000",     // commission
+    "ap": "1.3393"         // average price (체결가)
+  }
+}
+```
+
+### 판단 트리
+
+```
+msg["e"] == "ORDER_TRADE_UPDATE"?
+  └── order = msg["o"]
+      order["x"] == "TRADE" AND order["X"] == "FILLED"?
+        └── 청산 주문인가?
+            (order["R"] == true OR float(order["rp"]) != 0
+             OR order["o"] in {"TAKE_PROFIT_MARKET", "STOP_MARKET"})
+              ├── NO  → 무시 (진입 주문)
+              └── YES → close_reason 판별:
+                          "TAKE_PROFIT_MARKET" → "TP"
+                          "STOP_MARKET"        → "SL"
+                          그 외               → "MANUAL"
+                        net_pnl = float(rp) - abs(float(n))
+                        exit_price = float(order["ap"])
+                        await on_order_filled(net_pnl, close_reason, exit_price)
+```
+
+---
+
+## 섹션 3: `_on_position_closed()` 콜백 (`src/bot.py`)
+
+### 진입가 상태 저장
+
+`_open_position()` 내부에서 진입가와 수량을 인스턴스 변수로 저장한다. 청산 시점에는 포지션이 이미 사라져 있으므로 사전 저장이 필수다.
+
+```python
+# __init__에 추가
+self._entry_price: float | None = None
+self._entry_quantity: float | None = None
+
+# _open_position() 내부에서 저장
+self._entry_price = price
+self._entry_quantity = quantity
+```
+
+### 예상 PnL 계산
+
+```python
+def _calc_estimated_pnl(self, exit_price: float) -> float:
+    if self._entry_price is None or self._entry_quantity is None:
+        return 0.0
+    if self.current_trade_side == "LONG":
+        return (exit_price - self._entry_price) * self._entry_quantity
+    else:  # SHORT
+        return (self._entry_price - exit_price) * self._entry_quantity
+```
+
+### 콜백 전체 흐름
+
+```python
+async def _on_position_closed(
+    self,
+    net_pnl: float,
+    close_reason: str,   # "TP" | "SL" | "MANUAL"
+    exit_price: float,
+):
+    estimated_pnl = self._calc_estimated_pnl(exit_price)
+    diff = net_pnl - estimated_pnl  # 슬리피지 + 수수료 차이
+
+    # RiskManager에 순수익 기록
+    self.risk.record_pnl(net_pnl)
+
+    # Discord 알림
+    self.notifier.notify_close(
+        symbol=self.config.symbol,
+        side=self.current_trade_side or "UNKNOWN",
+        close_reason=close_reason,
+        exit_price=exit_price,
+        estimated_pnl=estimated_pnl,
+        net_pnl=net_pnl,
+        diff=diff,
+    )
+
+    logger.success(
+        f"포지션 청산({close_reason}): 예상={estimated_pnl:+.4f}, "
+        f"순수익={net_pnl:+.4f}, 차이={diff:+.4f} USDT"
+    )
+
+    # 봇 상태 초기화 (Flat 상태로 복귀)
+    self.current_trade_side = None
+    self._entry_price = None
+    self._entry_quantity = None
+```
+
+### 기존 `_close_position()` 변경
+
+봇이 직접 청산하는 경우(`_close_and_reenter`)에도 User Data Stream의 `ORDER_TRADE_UPDATE`가 발생한다. **중복 처리 방지**를 위해 `_close_position()`에서 `notify_close()`와 `record_pnl()` 호출을 제거한다. 모든 청산 후처리는 `_on_position_closed()` 콜백 하나로 일원화한다.
+
+---
+
+## 섹션 4: Discord 알림 포맷 (`src/notifier.py`)
+
+### `notify_close()` 시그니처 변경
+
+```python
+def notify_close(
+    self,
+    symbol: str,
+    side: str,
+    close_reason: str,    # "TP" | "SL" | "MANUAL"
+    exit_price: float,
+    estimated_pnl: float,
+    net_pnl: float,
+    diff: float,          # net_pnl - estimated_pnl
+) -> None:
+```
+
+### 알림 포맷
+
+```
+✅ [XRPUSDT] SHORT TP 청산
+청산가:          `1.3393`
+예상 수익:       `+0.4821 USDT`
+실제 순수익:     `+0.4612 USDT`
+차이(슬리피지+수수료): `-0.0209 USDT`
+```
+
+| 청산 원인 | 이모지 |
+|----------|--------|
+| TP       | ✅     |
+| SL       | ❌     |
+| MANUAL   | 🔶     |
+
+---
+
+## 섹션 5: `src/exchange.py` 추가 메서드
+
+```python
+async def create_listen_key(self) -> str:
+    """POST /fapi/v1/listenKey — listenKey 신규 발급"""
+
+async def keepalive_listen_key(self, listen_key: str) -> None:
+    """PUT /fapi/v1/listenKey — listenKey 만료 연장"""
+
+async def delete_listen_key(self, listen_key: str) -> None:
+    """DELETE /fapi/v1/listenKey — listenKey 삭제 (정상 종료 시)"""
+```
+
+---
+
+## 데이터 흐름 요약
+
+```
+Binance WebSocket
+  → ORDER_TRADE_UPDATE (FILLED, reduceOnly)
+  → UserDataStream._handle_message()
+  → net_pnl = rp - |commission|
+  → bot._on_position_closed(net_pnl, close_reason, exit_price)
+      ├── estimated_pnl = (exit - entry) × qty  (봇 계산)
+      ├── diff = net_pnl - estimated_pnl
+      ├── risk.record_pnl(net_pnl)              → 일일 PnL 누적
+      ├── notifier.notify_close(...)             → Discord 알림
+      └── 상태 초기화 (current_trade_side, _entry_price, _entry_quantity = None)
+```
+
+---
+
+## 제외 범위 (YAGNI)
+
+- DB 영구 저장 (SQLite/Postgres) — 현재 로그 기반으로 충분
+- 진입 주문 체결 알림 (`TRADE` + not reduceOnly) — 기존 `notify_open()`으로 커버
+- 부분 청산(partial fill) 처리 — 현재 봇은 전량 청산만 사용

docs/plans/2026-03-02-user-data-stream-tp-sl-detection-plan.md

@@ -0,0 +1,510 @@
+# User Data Stream TP/SL 감지 Implementation Plan
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** Binance Futures User Data Stream을 도입하여 TP/SL 작동을 실시간 감지하고, 순수익(Net PnL)을 기록하며, Discord에 예상 수익 vs 실제 순수익 비교 알림을 전송한다.
+
+**Architecture:** `python-binance`의 `futures_user_socket(listenKey)`로 User Data Stream에 연결하고, 30분 keepalive 백그라운드 태스크와 `while True: try-except` 무한 재연결 루프로 안정성을 확보한다. `ORDER_TRADE_UPDATE` 이벤트에서 청산 주문을 감지하면 `bot._on_position_closed()` 콜백을 호출하여 PnL 기록과 Discord 알림을 일원화한다.
+
+**Tech Stack:** Python 3.12, python-binance (AsyncClient, BinanceSocketManager), asyncio, loguru
+
+**Design Doc:** `docs/plans/2026-03-02-user-data-stream-tp-sl-detection-design.md`
+
+---
+
+## Task 1: `exchange.py`에 listenKey 관리 메서드 추가
+
+**Files:**
+- Modify: `src/exchange.py` (끝에 메서드 추가)
+
+**Step 1: listenKey 3개 메서드 구현**
+
+`src/exchange.py` 끝에 아래 메서드 3개를 추가한다.
+
+```python
+async def create_listen_key(self) -> str:
+    """POST /fapi/v1/listenKey — listenKey 신규 발급"""
+    loop = asyncio.get_event_loop()
+    result = await loop.run_in_executor(
+        None,
+        lambda: self.client.futures_stream_get_listen_key(),
+    )
+    return result
+
+async def keepalive_listen_key(self, listen_key: str) -> None:
+    """PUT /fapi/v1/listenKey — listenKey 만료 연장 (60분 → 리셋)"""
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(
+        None,
+        lambda: self.client.futures_stream_keepalive(listenKey=listen_key),
+    )
+
+async def delete_listen_key(self, listen_key: str) -> None:
+    """DELETE /fapi/v1/listenKey — listenKey 삭제 (정상 종료 시)"""
+    loop = asyncio.get_event_loop()
+    try:
+        await loop.run_in_executor(
+            None,
+            lambda: self.client.futures_stream_close(listenKey=listen_key),
+        )
+    except Exception as e:
+        logger.warning(f"listenKey 삭제 실패 (무시): {e}")
+```
+
+**Step 2: 커밋**
+
+```bash
+git add src/exchange.py
+git commit -m "feat: add listenKey create/keepalive/delete methods to exchange"
+```
+
+---
+
+## Task 2: `notifier.py`의 `notify_close()` 시그니처 확장
+
+**Files:**
+- Modify: `src/notifier.py`
+
+**Step 1: `notify_close()` 메서드 교체**
+
+기존 `notify_close()`를 아래로 교체한다. `close_reason`, `estimated_pnl`, `net_pnl`, `diff` 파라미터가 추가된다.
+
+```python
+def notify_close(
+    self,
+    symbol: str,
+    side: str,
+    close_reason: str,      # "TP" | "SL" | "MANUAL"
+    exit_price: float,
+    estimated_pnl: float,   # 봇 계산 (entry-exit 기반)
+    net_pnl: float,         # 바이낸스 rp - |commission|
+    diff: float,            # net_pnl - estimated_pnl (슬리피지+수수료)
+) -> None:
+    emoji_map = {"TP": "✅", "SL": "❌", "MANUAL": "🔶"}
+    emoji = emoji_map.get(close_reason, "🔶")
+    msg = (
+        f"{emoji} **[{symbol}] {side} {close_reason} 청산**\n"
+        f"청산가:               `{exit_price:.4f}`\n"
+        f"예상 수익:            `{estimated_pnl:+.4f} USDT`\n"
+        f"실제 순수익:          `{net_pnl:+.4f} USDT`\n"
+        f"차이(슬리피지+수수료): `{diff:+.4f} USDT`"
+    )
+    self._send(msg)
+```
+
+**Step 2: 커밋**
+
+```bash
+git add src/notifier.py
+git commit -m "feat: extend notify_close with close_reason, net_pnl, diff fields"
+```
+
+---
+
+## Task 3: `src/user_data_stream.py` 신규 생성
+
+**Files:**
+- Create: `src/user_data_stream.py`
+
+**Step 1: 파일 전체 작성**
+
+```python
+import asyncio
+from typing import Callable
+from binance import AsyncClient, BinanceSocketManager
+from loguru import logger
+
+_KEEPALIVE_INTERVAL = 30 * 60   # 30분 (listenKey 만료 60분의 절반)
+_RECONNECT_DELAY    = 5         # 재연결 대기 초
+
+_CLOSE_ORDER_TYPES = {"TAKE_PROFIT_MARKET", "STOP_MARKET"}
+
+
+class UserDataStream:
+    """
+    Binance Futures User Data Stream을 구독하여 주문 체결 이벤트를 처리한다.
+
+    - listenKey 30분 keepalive 백그라운드 태스크
+    - 네트워크 단절 시 무한 재연결 루프
+    - ORDER_TRADE_UPDATE 이벤트에서 청산 주문만 필터링하여 콜백 호출
+    """
+
+    def __init__(
+        self,
+        exchange,                        # BinanceFuturesClient 인스턴스
+        on_order_filled: Callable,       # bot._on_position_closed 콜백
+    ):
+        self._exchange = exchange
+        self._on_order_filled = on_order_filled
+        self._listen_key: str | None = None
+        self._keepalive_task: asyncio.Task | None = None
+
+    async def start(self, api_key: str, api_secret: str) -> None:
+        """User Data Stream 메인 루프 — 봇 종료 시까지 실행."""
+        client = await AsyncClient.create(
+            api_key=api_key,
+            api_secret=api_secret,
+        )
+        bm = BinanceSocketManager(client)
+        try:
+            await self._run_loop(bm)
+        finally:
+            await client.close_connection()
+
+    async def _run_loop(self, bm: BinanceSocketManager) -> None:
+        """listenKey 발급 → 연결 → 재연결 무한 루프."""
+        while True:
+            try:
+                self._listen_key = await self._exchange.create_listen_key()
+                logger.info(f"User Data Stream listenKey 발급: {self._listen_key[:8]}...")
+
+                self._keepalive_task = asyncio.create_task(
+                    self._keepalive_loop(self._listen_key)
+                )
+
+                async with bm.futures_user_socket(self._listen_key) as stream:
+                    logger.info("User Data Stream 연결 완료")
+                    async for msg in stream:
+                        await self._handle_message(msg)
+
+            except asyncio.CancelledError:
+                logger.info("User Data Stream 정상 종료")
+                if self._listen_key:
+                    await self._exchange.delete_listen_key(self._listen_key)
+                if self._keepalive_task:
+                    self._keepalive_task.cancel()
+                break
+
+            except Exception as e:
+                logger.warning(
+                    f"User Data Stream 끊김: {e} — "
+                    f"{_RECONNECT_DELAY}초 후 재연결"
+                )
+                if self._keepalive_task:
+                    self._keepalive_task.cancel()
+                    self._keepalive_task = None
+                await asyncio.sleep(_RECONNECT_DELAY)
+
+    async def _keepalive_loop(self, listen_key: str) -> None:
+        """30분마다 listenKey를 갱신한다."""
+        while True:
+            await asyncio.sleep(_KEEPALIVE_INTERVAL)
+            try:
+                await self._exchange.keepalive_listen_key(listen_key)
+                logger.debug("listenKey 갱신 완료")
+            except Exception as e:
+                logger.warning(f"listenKey 갱신 실패: {e} — 재연결 루프가 처리")
+                break
+
+    async def _handle_message(self, msg: dict) -> None:
+        """ORDER_TRADE_UPDATE 이벤트에서 청산 주문을 필터링하여 콜백을 호출한다."""
+        if msg.get("e") != "ORDER_TRADE_UPDATE":
+            return
+
+        order = msg.get("o", {})
+
+        # x: Execution Type, X: Order Status
+        if order.get("x") != "TRADE" or order.get("X") != "FILLED":
+            return
+
+        order_type   = order.get("o", "")
+        is_reduce    = order.get("R", False)
+        realized_pnl = float(order.get("rp", "0"))
+
+        # 청산 주문 판별: reduceOnly이거나, TP/SL 타입이거나, rp != 0
+        is_close = is_reduce or order_type in _CLOSE_ORDER_TYPES or realized_pnl != 0
+        if not is_close:
+            return
+
+        commission = abs(float(order.get("n", "0")))
+        net_pnl    = realized_pnl - commission
+        exit_price = float(order.get("ap", "0"))
+
+        if order_type == "TAKE_PROFIT_MARKET":
+            close_reason = "TP"
+        elif order_type == "STOP_MARKET":
+            close_reason = "SL"
+        else:
+            close_reason = "MANUAL"
+
+        logger.info(
+            f"청산 감지({close_reason}): exit={exit_price:.4f}, "
+            f"rp={realized_pnl:+.4f}, commission={commission:.4f}, "
+            f"net_pnl={net_pnl:+.4f}"
+        )
+
+        await self._on_order_filled(
+            net_pnl=net_pnl,
+            close_reason=close_reason,
+            exit_price=exit_price,
+        )
+```
+
+**Step 2: 커밋**
+
+```bash
+git add src/user_data_stream.py
+git commit -m "feat: add UserDataStream with keepalive and reconnect loop"
+```
+
+---
+
+## Task 4: `bot.py` 수정 — 상태 변수 추가 및 `_open_position()` 저장
+
+**Files:**
+- Modify: `src/bot.py`
+
+**Step 1: `__init__`에 상태 변수 추가**
+
+`TradingBot.__init__()` 내부에서 `self.current_trade_side` 선언 바로 아래에 추가한다.
+
+```python
+self._entry_price: float | None = None
+self._entry_quantity: float | None = None
+```
+
+**Step 2: `_open_position()` 내부에서 진입가/수량 저장**
+
+`self.current_trade_side = signal` 바로 아래에 추가한다.
+
+```python
+self._entry_price = price
+self._entry_quantity = quantity
+```
+
+**Step 3: 커밋**
+
+```bash
+git add src/bot.py
+git commit -m "feat: store entry_price and entry_quantity on position open"
+```
+
+---
+
+## Task 5: `bot.py` 수정 — `_on_position_closed()` 콜백 추가
+
+**Files:**
+- Modify: `src/bot.py`
+
+**Step 1: `_calc_estimated_pnl()` 헬퍼 메서드 추가**
+
+`_close_position()` 메서드 바로 위에 추가한다.
+
+```python
+def _calc_estimated_pnl(self, exit_price: float) -> float:
+    """진입가·수량 기반 예상 PnL 계산 (수수료 미반영)."""
+    if self._entry_price is None or self._entry_quantity is None:
+        return 0.0
+    if self.current_trade_side == "LONG":
+        return (exit_price - self._entry_price) * self._entry_quantity
+    return (self._entry_price - exit_price) * self._entry_quantity
+```
+
+**Step 2: `_on_position_closed()` 콜백 추가**
+
+`_calc_estimated_pnl()` 바로 아래에 추가한다.
+
+```python
+async def _on_position_closed(
+    self,
+    net_pnl: float,
+    close_reason: str,
+    exit_price: float,
+) -> None:
+    """User Data Stream에서 청산 감지 시 호출되는 콜백."""
+    estimated_pnl = self._calc_estimated_pnl(exit_price)
+    diff = net_pnl - estimated_pnl
+
+    self.risk.record_pnl(net_pnl)
+
+    self.notifier.notify_close(
+        symbol=self.config.symbol,
+        side=self.current_trade_side or "UNKNOWN",
+        close_reason=close_reason,
+        exit_price=exit_price,
+        estimated_pnl=estimated_pnl,
+        net_pnl=net_pnl,
+        diff=diff,
+    )
+
+    logger.success(
+        f"포지션 청산({close_reason}): 예상={estimated_pnl:+.4f}, "
+        f"순수익={net_pnl:+.4f}, 차이={diff:+.4f} USDT"
+    )
+
+    # Flat 상태로 초기화
+    self.current_trade_side = None
+    self._entry_price = None
+    self._entry_quantity = None
+```
+
+**Step 3: 커밋**
+
+```bash
+git add src/bot.py
+git commit -m "feat: add _on_position_closed callback with net PnL and discord alert"
+```
+
+---
+
+## Task 6: `bot.py` 수정 — `_close_position()`에서 중복 후처리 제거
+
+**Files:**
+- Modify: `src/bot.py`
+
+**배경:** 봇이 직접 청산(`_close_and_reenter`)하는 경우에도 User Data Stream의 `ORDER_TRADE_UPDATE`가 발생한다. 중복 방지를 위해 `_close_position()`에서 `notify_close()`와 `record_pnl()` 호출을 제거한다.
+
+**Step 1: `_close_position()` 수정**
+
+기존 코드:
+```python
+async def _close_position(self, position: dict):
+    amt = abs(float(position["positionAmt"]))
+    side = "SELL" if float(position["positionAmt"]) > 0 else "BUY"
+    pos_side = "LONG" if side == "SELL" else "SHORT"
+    await self.exchange.cancel_all_orders()
+    await self.exchange.place_order(side=side, quantity=amt, reduce_only=True)
+
+    entry = float(position["entryPrice"])
+    mark  = float(position["markPrice"])
+    pnl   = (mark - entry) * amt if side == "SELL" else (entry - mark) * amt
+
+    self.notifier.notify_close(
+        symbol=self.config.symbol,
+        side=pos_side,
+        exit_price=mark,
+        pnl=pnl,
+    )
+    self.risk.record_pnl(pnl)
+    self.current_trade_side = None
+    logger.success(f"포지션 청산: PnL={pnl:.4f} USDT")
+```
+
+수정 후 (`notify_close`, `record_pnl`, `current_trade_side = None` 제거 — User Data Stream 콜백이 처리):
+```python
+async def _close_position(self, position: dict):
+    """포지션 청산 주문만 실행한다. PnL 기록/알림은 _on_position_closed 콜백이 담당."""
+    amt = abs(float(position["positionAmt"]))
+    side = "SELL" if float(position["positionAmt"]) > 0 else "BUY"
+    await self.exchange.cancel_all_orders()
+    await self.exchange.place_order(side=side, quantity=amt, reduce_only=True)
+    logger.info(f"청산 주문 전송 완료 (side={side}, qty={amt})")
+```
+
+**Step 2: 커밋**
+
+```bash
+git add src/bot.py
+git commit -m "refactor: remove duplicate pnl/notify from _close_position (handled by callback)"
+```
+
+---
+
+## Task 7: `bot.py` 수정 — `run()`에서 UserDataStream 병렬 실행
+
+**Files:**
+- Modify: `src/bot.py`
+
+**Step 1: import 추가**
+
+파일 상단 import 블록에 추가한다.
+
+```python
+from src.user_data_stream import UserDataStream
+```
+
+**Step 2: `run()` 메서드 수정**
+
+기존:
+```python
+async def run(self):
+    logger.info(f"봇 시작: {self.config.symbol}, 레버리지 {self.config.leverage}x")
+    await self._recover_position()
+    balance = await self.exchange.get_balance()
+    self.risk.set_base_balance(balance)
+    logger.info(f"기준 잔고 설정: {balance:.2f} USDT (동적 증거금 비율 기준점)")
+    await self.stream.start(
+        api_key=self.config.api_key,
+        api_secret=self.config.api_secret,
+    )
+```
+
+수정 후:
+```python
+async def run(self):
+    logger.info(f"봇 시작: {self.config.symbol}, 레버리지 {self.config.leverage}x")
+    await self._recover_position()
+    balance = await self.exchange.get_balance()
+    self.risk.set_base_balance(balance)
+    logger.info(f"기준 잔고 설정: {balance:.2f} USDT (동적 증거금 비율 기준점)")
+
+    user_stream = UserDataStream(
+        exchange=self.exchange,
+        on_order_filled=self._on_position_closed,
+    )
+
+    await asyncio.gather(
+        self.stream.start(
+            api_key=self.config.api_key,
+            api_secret=self.config.api_secret,
+        ),
+        user_stream.start(
+            api_key=self.config.api_key,
+            api_secret=self.config.api_secret,
+        ),
+    )
+```
+
+**Step 3: 커밋**
+
+```bash
+git add src/bot.py
+git commit -m "feat: run UserDataStream in parallel with candle stream"
+```
+
+---
+
+## Task 8: README.md 업데이트
+
+**Files:**
+- Modify: `README.md`
+
+**Step 1: 기능 목록에 User Data Stream 항목 추가**
+
+README의 주요 기능 섹션에 아래 내용을 추가한다.
+
+- **실시간 TP/SL 감지**: Binance User Data Stream으로 TP/SL 작동을 즉시 감지 (캔들 마감 대기 없음)
+- **순수익(Net PnL) 기록**: 바이낸스 `realizedProfit - commission`으로 정확한 순수익 계산
+- **Discord 상세 청산 알림**: 예상 수익 vs 실제 순수익 + 슬리피지/수수료 차이 표시
+- **listenKey 자동 갱신**: 30분 keepalive + 네트워크 단절 시 자동 재연결
+
+**Step 2: 커밋**
+
+```bash
+git add README.md
+git commit -m "docs: update README with User Data Stream TP/SL detection feature"
+```
+
+---
+
+## 최종 검증
+
+봇 실행 후 로그에서 아래 메시지가 순서대로 나타나면 정상 동작:
+
+```
+INFO  | User Data Stream listenKey 발급: xxxxxxxx...
+INFO  | User Data Stream 연결 완료
+DEBUG | listenKey 갱신 완료  ← 30분 후
+INFO  | 청산 감지(TP): exit=1.3393, rp=+0.4821, commission=0.0209, net_pnl=+0.4612
+SUCCESS | 포지션 청산(TP): 예상=+0.4821, 순수익=+0.4612, 차이=-0.0209 USDT
+```
+
+Discord에는 아래 형식의 알림이 전송됨:
+
+```
+✅ [XRPUSDT] SHORT TP 청산
+청산가:               1.3393
+예상 수익:            +0.4821 USDT
+실제 순수익:          +0.4612 USDT
+차이(슬리피지+수수료): -0.0209 USDT
+```

models/training_log.json

@@ -216,5 +216,90 @@
     "train_sec": 0.1,
     "time_weight_decay": 2.0,
     "model_path": "models/mlx_filter.weights"
+  },
+  {
+    "date": "2026-03-01T23:59:27.956019",
+    "backend": "mlx",
+    "auc": 0.5595,
+    "best_threshold": 0.9538,
+    "best_precision": 0.462,
+    "best_recall": 0.171,
+    "samples": 533,
+    "train_sec": 0.2,
+    "time_weight_decay": 2.0,
+    "model_path": "models/mlx_filter.weights"
+  },
+  {
+    "date": "2026-03-02T00:40:15.931055",
+    "backend": "mlx",
+    "auc": 0.5829,
+    "best_threshold": 0.9609,
+    "best_precision": 0.6,
+    "best_recall": 0.171,
+    "samples": 534,
+    "train_sec": 0.2,
+    "time_weight_decay": 2.0,
+    "model_path": "models/mlx_filter.weights"
+  },
+  {
+    "date": "2026-03-02T00:54:32.264425",
+    "backend": "lgbm",
+    "auc": 0.5607,
+    "best_threshold": 0.6532,
+    "best_precision": 0.467,
+    "best_recall": 0.2,
+    "samples": 533,
+    "features": 23,
+    "time_weight_decay": 2.0,
+    "model_path": "models/lgbm_filter.pkl"
+  },
+  {
+    "date": "2026-03-02T01:07:30.690959",
+    "backend": "lgbm",
+    "auc": 0.5579,
+    "best_threshold": 0.6511,
+    "best_precision": 0.4,
+    "best_recall": 0.171,
+    "samples": 533,
+    "features": 23,
+    "time_weight_decay": 2.0,
+    "model_path": "models/lgbm_filter.pkl"
+  },
+  {
+    "date": "2026-03-02T02:00:45.931227",
+    "backend": "lgbm",
+    "auc": 0.5752,
+    "best_threshold": 0.6307,
+    "best_precision": 0.471,
+    "best_recall": 0.229,
+    "samples": 533,
+    "features": 23,
+    "time_weight_decay": 2.0,
+    "model_path": "models/lgbm_filter.pkl"
+  },
+  {
+    "date": "2026-03-02T14:51:09.101738",
+    "backend": "lgbm",
+    "auc": 0.5361,
+    "best_threshold": 0.5308,
+    "best_precision": 0.406,
+    "best_recall": 0.371,
+    "samples": 533,
+    "features": 23,
+    "time_weight_decay": 2.0,
+    "model_path": "models/lgbm_filter.pkl",
+    "tuned_params_path": "models/tune_results_20260302_144749.json",
+    "lgbm_params": {
+      "n_estimators": 434,
+      "learning_rate": 0.123659,
+      "num_leaves": 14,
+      "min_child_samples": 10,
+      "subsample": 0.929062,
+      "colsample_bytree": 0.94633,
+      "reg_alpha": 0.573971,
+      "reg_lambda": 0.000157,
+      "max_depth": 6
+    },
+    "weight_scale": 1.783105
   }
 ]

requirements.txt

@@ -1,4 +1,4 @@
-python-binance==1.0.19
+python-binance>=1.0.28
 pandas>=2.3.2
 pandas-ta==0.4.71b0
 python-dotenv==1.0.0
@@ -13,3 +13,4 @@ scikit-learn>=1.4.0
 joblib>=1.3.0
 pyarrow>=15.0.0
 onnxruntime>=1.18.0
+optuna>=3.6.0

scripts/deploy_model.sh

@@ -33,12 +33,25 @@ done
 echo "=== 모델 전송 시작 (백엔드: ${BACKEND}) ==="
 echo "  대상: ${LXC_HOST}:${LXC_MODELS_PATH}"
 
-# ── 원격 디렉터리 생성 + lgbm 기존 모델 백업 ─────────────────────────────────
+# ── 원격 디렉터리 생성 + 백업 + 상대 백엔드 파일 제거 ───────────────────────
+# lgbm 배포 시: 기존 lgbm 백업 후 ONNX 파일 삭제 (ONNX 우선순위 때문에 lgbm이 무시되는 것 방지)
+# mlx 배포 시: lgbm 파일 삭제 (명시적으로 mlx만 사용)
 ssh "${LXC_HOST}" "
   mkdir -p '${LXC_MODELS_PATH}'
-  if [ '$BACKEND' = 'lgbm' ] && [ -f '${LXC_MODELS_PATH}/lgbm_filter.pkl' ]; then
-    cp '${LXC_MODELS_PATH}/lgbm_filter.pkl' '${LXC_MODELS_PATH}/lgbm_filter_prev.pkl'
-    echo '  기존 lgbm 모델 백업 완료'
+  if [ '$BACKEND' = 'lgbm' ]; then
+    if [ -f '${LXC_MODELS_PATH}/lgbm_filter.pkl' ]; then
+      cp '${LXC_MODELS_PATH}/lgbm_filter.pkl' '${LXC_MODELS_PATH}/lgbm_filter_prev.pkl'
+      echo '  기존 lgbm 모델 백업 완료'
+    fi
+    if [ -f '${LXC_MODELS_PATH}/mlx_filter.weights.onnx' ]; then
+      rm '${LXC_MODELS_PATH}/mlx_filter.weights.onnx'
+      echo '  ONNX 파일 제거 완료 (lgbm 우선 적용)'
+    fi
+  else
+    if [ -f '${LXC_MODELS_PATH}/lgbm_filter.pkl' ]; then
+      rm '${LXC_MODELS_PATH}/lgbm_filter.pkl'
+      echo '  lgbm 파일 제거 완료 (mlx 우선 적용)'
+    fi
   fi
 "
 

scripts/fetch_history.py

@@ -259,6 +259,61 @@ async def _fetch_oi_and_funding(
     return _merge_oi_funding(candles, oi_df, funding_df)
 
 
+def upsert_parquet(path: "Path | str", new_df: pd.DataFrame) -> pd.DataFrame:
+    """
+    기존 parquet 파일에 신규 데이터를 Upsert(병합)한다.
+
+    규칙:
+    - 기존 행의 oi_change / funding_rate가 0.0이면 신규 값으로 덮어씀
+    - 기존 행의 oi_change / funding_rate가 이미 0이 아니면 유지
+    - 신규 타임스탬프 행은 그냥 추가
+    - 결과는 timestamp 기준 오름차순 정렬, 중복 제거
+
+    Args:
+        path: 기존 parquet 경로 (없으면 new_df 그대로 반환)
+        new_df: 새로 수집한 DataFrame (timestamp index)
+
+    Returns:
+        병합된 DataFrame
+    """
+    path = Path(path)
+    if not path.exists():
+        return new_df.sort_index()
+
+    existing = pd.read_parquet(path)
+
+    # timestamp index 통일 (tz-aware UTC)
+    if existing.index.tz is None:
+        existing.index = existing.index.tz_localize("UTC")
+    if new_df.index.tz is None:
+        new_df.index = new_df.index.tz_localize("UTC")
+
+    # 기존 데이터에서 oi_change / funding_rate가 0.0인 행만 신규 값으로 업데이트
+    UPSERT_COLS = ["oi_change", "funding_rate"]
+    overlap_idx = existing.index.intersection(new_df.index)
+
+    for col in UPSERT_COLS:
+        if col not in existing.columns or col not in new_df.columns:
+            continue
+        # 겹치는 행 중 기존 값이 0.0인 경우에만 신규 값으로 교체
+        zero_mask = existing.loc[overlap_idx, col] == 0.0
+        update_idx = overlap_idx[zero_mask]
+        if len(update_idx) > 0:
+            existing.loc[update_idx, col] = new_df.loc[update_idx, col]
+
+    # 신규 타임스탬프 행 추가 (기존에 없는 것만)
+    new_only_idx = new_df.index.difference(existing.index)
+    if len(new_only_idx) > 0:
+        existing = pd.concat([existing, new_df.loc[new_only_idx]])
+
+    # 컬럼 불일치(기존 parquet에 oi_change/funding_rate 없음)로 생긴 NaN을 0으로 채움
+    for col in UPSERT_COLS:
+        if col in existing.columns:
+            existing[col] = existing[col].fillna(0.0)
+
+    return existing.sort_index()
+
+
 def main():
     parser = argparse.ArgumentParser(
         description="바이낸스 선물 과거 캔들 수집. 단일 심볼 또는 멀티 심볼 병합 저장."
@@ -272,6 +327,10 @@ def main():
         "--no-oi", action="store_true",
         help="OI/펀딩비 수집을 건너뜀 (캔들 데이터만 저장)",
     )
+    parser.add_argument(
+        "--no-upsert", action="store_true",
+        help="기존 parquet을 Upsert하지 않고 새로 덮어씀 (기본: Upsert 활성화)",
+    )
     args = parser.parse_args()
 
     # 하위 호환: --symbol 단독 사용 시 symbols로 통합
@@ -283,8 +342,10 @@ def main():
         if not args.no_oi:
             print(f"\n[OI/펀딩비] {args.symbols[0]} 수집 중...")
             df = asyncio.run(_fetch_oi_and_funding(args.symbols[0], args.days, df))
+        if not args.no_upsert:
+            df = upsert_parquet(args.output, df)
         df.to_parquet(args.output)
-        print(f"저장 완료: {args.output} ({len(df):,}행, {len(df.columns)}컬럼)")
+        print(f"{'Upsert' if not args.no_upsert else '저장'} 완료: {args.output} ({len(df):,}행, {len(df.columns)}컬럼)")
     else:
         # 멀티 심볼: 단일 클라이언트로 순차 수집 후 타임스탬프 기준 inner join 병합
         dfs = asyncio.run(fetch_klines_all(args.symbols, args.interval, args.days))
@@ -304,8 +365,10 @@ def main():
             merged = asyncio.run(_fetch_oi_and_funding(primary, args.days, merged))
 
         output = args.output.replace("xrpusdt", "combined")
+        if not args.no_upsert:
+            merged = upsert_parquet(output, merged)
         merged.to_parquet(output)
-        print(f"\n병합 저장 완료: {output} ({len(merged):,}행, {len(merged.columns)}컬럼)")
+        print(f"\n{'Upsert' if not args.no_upsert else '병합 저장'} 완료: {output} ({len(merged):,}행, {len(merged.columns)}컬럼)")
 
 
 if __name__ == "__main__":

scripts/run_optuna.sh

@@ -0,0 +1,49 @@
+#!/usr/bin/env bash
+# Optuna로 LightGBM 하이퍼파라미터를 탐색하고 결과를 출력한다.
+# 사람이 결과를 확인·승인한 후 train_model.py에 수동으로 반영하는 방식.
+#
+# 사용법:
+#   bash scripts/run_optuna.sh              # 기본 (50 trials, 5폴드)
+#   bash scripts/run_optuna.sh 100          # 100 trials
+#   bash scripts/run_optuna.sh 100 3        # 100 trials, 3폴드
+#   bash scripts/run_optuna.sh 10 3 --no-baseline  # 빠른 테스트
+#
+# 결과 확인 후 승인하면:
+#   python scripts/train_model.py --tuned-params models/tune_results_YYYYMMDD_HHMMSS.json
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+VENV_PATH="${VENV_PATH:-$PROJECT_ROOT/.venv}"
+if [ -f "$VENV_PATH/bin/activate" ]; then
+    # shellcheck source=/dev/null
+    source "$VENV_PATH/bin/activate"
+else
+    echo "경고: 가상환경을 찾을 수 없습니다 ($VENV_PATH). 시스템 Python을 사용합니다." >&2
+fi
+
+TRIALS="${1:-50}"
+FOLDS="${2:-5}"
+EXTRA_ARGS="${3:-}"
+
+cd "$PROJECT_ROOT"
+
+echo "=== Optuna 하이퍼파라미터 탐색 ==="
+echo "  trials=${TRIALS}, folds=${FOLDS}"
+echo ""
+
+python scripts/tune_hyperparams.py \
+    --trials "$TRIALS" \
+    --folds  "$FOLDS" \
+    $EXTRA_ARGS
+
+echo ""
+echo "=== 탐색 완료 ==="
+echo ""
+echo "결과 JSON을 확인하고 승인하면 아래 명령으로 재학습하세요:"
+echo "  python scripts/train_model.py --tuned-params models/tune_results_<timestamp>.json"
+echo ""
+echo "Walk-Forward 검증과 함께 재학습:"
+echo "  python scripts/train_model.py --tuned-params models/tune_results_<timestamp>.json --wf"

scripts/run_tests.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+# 전체 테스트 실행 스크립트
+#
+# 사용법:
+#   bash scripts/run_tests.sh           # 전체 실행
+#   bash scripts/run_tests.sh -k bot    # 특정 키워드 필터
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+VENV_PATH="${VENV_PATH:-$PROJECT_ROOT/.venv}"
+if [ -f "$VENV_PATH/bin/activate" ]; then
+    # shellcheck source=/dev/null
+    source "$VENV_PATH/bin/activate"
+else
+    echo "경고: 가상환경을 찾을 수 없습니다 ($VENV_PATH). 시스템 Python을 사용합니다." >&2
+fi
+
+cd "$PROJECT_ROOT"
+
+python -m pytest tests/ \
+    --ignore=tests/test_database.py \
+    -v \
+    "$@"

scripts/test_reverse_reenter.sh

@@ -0,0 +1,83 @@
+#!/usr/bin/env bash
+# 반대 시그널 재진입 기능 테스트 스크립트
+# 사용법: bash scripts/test_reverse_reenter.sh [task]
+#
+# 예시:
+#   bash scripts/test_reverse_reenter.sh          # 전체 태스크 순서대로 실행
+#   bash scripts/test_reverse_reenter.sh 1        # Task 1: 신규 테스트만 (실패 확인)
+#   bash scripts/test_reverse_reenter.sh 2        # Task 2: _close_and_reenter 메서드 테스트
+#   bash scripts/test_reverse_reenter.sh 3        # Task 3: process_candle 분기 테스트
+#   bash scripts/test_reverse_reenter.sh bot      # test_bot.py 전체
+#   bash scripts/test_reverse_reenter.sh all      # tests/ 전체
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+VENV_PATH="${VENV_PATH:-$PROJECT_ROOT/.venv}"
+if [ -f "$VENV_PATH/bin/activate" ]; then
+    # shellcheck source=/dev/null
+    source "$VENV_PATH/bin/activate"
+else
+    echo "경고: 가상환경을 찾을 수 없습니다 ($VENV_PATH). 시스템 Python을 사용합니다." >&2
+fi
+
+cd "$PROJECT_ROOT"
+
+TASK="${1:-all}"
+
+# ── 태스크별 테스트 이름 ──────────────────────────────────────────────────────
+TASK1_TESTS=(
+    "tests/test_bot.py::test_close_and_reenter_calls_open_when_ml_passes"
+    "tests/test_bot.py::test_close_and_reenter_skips_open_when_ml_blocks"
+    "tests/test_bot.py::test_close_and_reenter_skips_open_when_max_positions_reached"
+)
+
+TASK2_TESTS=(
+    "tests/test_bot.py::test_close_and_reenter_calls_open_when_ml_passes"
+    "tests/test_bot.py::test_close_and_reenter_skips_open_when_ml_blocks"
+    "tests/test_bot.py::test_close_and_reenter_skips_open_when_max_positions_reached"
+)
+
+TASK3_TESTS=(
+    "tests/test_bot.py::test_process_candle_calls_close_and_reenter_on_reverse_signal"
+)
+
+run_pytest() {
+    echo ""
+    echo "▶ pytest $*"
+    echo "────────────────────────────────────────"
+    python -m pytest "$@" -v
+}
+
+case "$TASK" in
+    1)
+        echo "=== Task 1: 신규 테스트 실행 (구현 전 → FAIL 예상) ==="
+        run_pytest "${TASK1_TESTS[@]}"
+        ;;
+    2)
+        echo "=== Task 2: _close_and_reenter 메서드 테스트 (구현 후 → PASS 예상) ==="
+        run_pytest "${TASK2_TESTS[@]}"
+        ;;
+    3)
+        echo "=== Task 3: process_candle 분기 테스트 (수정 후 → PASS 예상) ==="
+        run_pytest "${TASK3_TESTS[@]}"
+        ;;
+    bot)
+        echo "=== test_bot.py 전체 ==="
+        run_pytest tests/test_bot.py
+        ;;
+    all)
+        echo "=== 전체 테스트 스위트 ==="
+        run_pytest tests/
+        ;;
+    *)
+        echo "알 수 없는 태스크: $TASK"
+        echo "사용법: bash scripts/test_reverse_reenter.sh [1|2|3|bot|all]"
+        exit 1
+        ;;
+esac
+
+echo ""
+echo "=== 완료 ==="

scripts/train_and_deploy.sh

@@ -27,12 +27,27 @@ WF_SPLITS="${2:-5}"   # 두 번째 인자: Walk-Forward 폴드 수 (0이면 건
 
 cd "$PROJECT_ROOT"
 
-echo "=== [1/3] 데이터 수집 (XRP + BTC + ETH 3심볼, 1년치 + OI/펀딩비) ==="
+mkdir -p data
+
+PARQUET_FILE="data/combined_15m.parquet"
+
+echo "=== [1/3] 데이터 수집 (XRP + BTC + ETH 3심볼 + OI/펀딩비) ==="
+if [ ! -f "$PARQUET_FILE" ]; then
+    echo "  [최초 실행] 기존 데이터 없음 → 1년치(365일) 전체 수집 (--no-upsert)"
+    FETCH_DAYS=365
+    UPSERT_FLAG="--no-upsert"
+else
+    echo "  [일반 실행] 기존 데이터 존재 → 35일치 Upsert (OI/펀딩비 0.0 구간 보충)"
+    FETCH_DAYS=35
+    UPSERT_FLAG=""
+fi
+
 python scripts/fetch_history.py \
     --symbols XRPUSDT BTCUSDT ETHUSDT \
     --interval 15m \
-    --days 365 \
-    --output data/combined_15m.parquet
+    --days "$FETCH_DAYS" \
+    $UPSERT_FLAG \
+    --output "$PARQUET_FILE"
 
 echo ""
 echo "=== [2/3] 모델 학습 (23개 피처: XRP 13 + BTC/ETH 8 + OI/펀딩비 2) ==="

scripts/train_model.py

@@ -146,7 +146,52 @@ def generate_dataset(df: pd.DataFrame, n_jobs: int | None = None) -> pd.DataFram
     return pd.DataFrame(rows)
 
 
-def train(data_path: str, time_weight_decay: float = 2.0):
+ACTIVE_PARAMS_PATH = Path("models/active_lgbm_params.json")
+
+
+def _load_lgbm_params(tuned_params_path: str | None) -> tuple[dict, float]:
+    """기본 LightGBM 파라미터를 반환하고, 튜닝 JSON이 주어지면 덮어쓴다.
+
+    우선순위:
+      1. --tuned-params 명시적 인자
+      2. models/active_lgbm_params.json (Optuna가 자동 갱신)
+      3. 코드 내 하드코딩 기본값 (fallback)
+    """
+    lgbm_params: dict = {
+        "n_estimators":      434,
+        "learning_rate":     0.123659,
+        "max_depth":         6,
+        "num_leaves":        14,
+        "min_child_samples": 10,
+        "subsample":         0.929062,
+        "colsample_bytree":  0.946330,
+        "reg_alpha":         0.573971,
+        "reg_lambda":        0.000157,
+    }
+    weight_scale = 1.783105
+
+    # 명시적 인자가 없으면 active 파일 자동 탐색
+    resolved_path = tuned_params_path or (
+        str(ACTIVE_PARAMS_PATH) if ACTIVE_PARAMS_PATH.exists() else None
+    )
+
+    if resolved_path:
+        with open(resolved_path, "r", encoding="utf-8") as f:
+            tune_data = json.load(f)
+        best_params = dict(tune_data["best_trial"]["params"])
+        weight_scale = float(best_params.pop("weight_scale", 1.0))
+        lgbm_params.update(best_params)
+        source = "명시적 인자" if tuned_params_path else "active 파일 자동 로드"
+        print(f"\n[Optuna] 튜닝 파라미터 로드 ({source}): {resolved_path}")
+        print(f"[Optuna] 적용 파라미터: {lgbm_params}")
+        print(f"[Optuna] weight_scale: {weight_scale}\n")
+    else:
+        print("[Optuna] active 파일 없음 → 코드 내 기본 파라미터 사용\n")
+
+    return lgbm_params, weight_scale
+
+
+def train(data_path: str, time_weight_decay: float = 2.0, tuned_params_path: str | None = None):
     print(f"데이터 로드: {data_path}")
     df_raw = pd.read_parquet(data_path)
     print(f"캔들 수: {len(df_raw)}, 컬럼: {list(df_raw.columns)}")
@@ -188,7 +233,10 @@ def train(data_path: str, time_weight_decay: float = 2.0):
     split = int(len(X) * 0.8)
     X_train, X_val = X.iloc[:split], X.iloc[split:]
     y_train, y_val = y.iloc[:split], y.iloc[split:]
-    w_train = w[:split]
+
+    # 튜닝 파라미터 로드 (없으면 기본값 사용)
+    lgbm_params, weight_scale = _load_lgbm_params(tuned_params_path)
+    w_train = (w[:split] * weight_scale).astype(np.float32)
 
     # --- 클래스 불균형 처리: 언더샘플링 (시간 가중치 인덱스 보존) ---
     pos_idx = np.where(y_train == 1)[0]
@@ -208,18 +256,7 @@ def train(data_path: str, time_weight_decay: float = 2.0):
     print(f"검증 데이터: {len(X_val)}개 (양성={int(y_val.sum())}, 음성={int((y_val==0).sum())})")
     # ---------------------------------------------------------------
 
-    model = lgb.LGBMClassifier(
-        n_estimators=500,
-        learning_rate=0.05,
-        num_leaves=31,
-        min_child_samples=15,
-        subsample=0.8,
-        colsample_bytree=0.8,
-        reg_alpha=0.05,
-        reg_lambda=0.1,
-        random_state=42,
-        verbose=-1,
-    )
+    model = lgb.LGBMClassifier(**lgbm_params, random_state=42, verbose=-1)
     model.fit(
         X_train, y_train,
         sample_weight=w_train,
@@ -268,7 +305,7 @@ def train(data_path: str, time_weight_decay: float = 2.0):
     if LOG_PATH.exists():
         with open(LOG_PATH) as f:
             log = json.load(f)
-    log.append({
+    log_entry: dict = {
         "date": datetime.now().isoformat(),
         "backend": "lgbm",
         "auc": round(auc, 4),
@@ -279,7 +316,11 @@ def train(data_path: str, time_weight_decay: float = 2.0):
         "features": len(actual_feature_cols),
         "time_weight_decay": time_weight_decay,
         "model_path": str(MODEL_PATH),
-    })
+        "tuned_params_path": tuned_params_path,
+        "lgbm_params": lgbm_params,
+        "weight_scale": weight_scale,
+    }
+    log.append(log_entry)
     with open(LOG_PATH, "w") as f:
         json.dump(log, f, indent=2)
 
@@ -291,6 +332,7 @@ def walk_forward_auc(
     time_weight_decay: float = 2.0,
     n_splits: int = 5,
     train_ratio: float = 0.6,
+    tuned_params_path: str | None = None,
 ) -> None:
     """Walk-Forward 검증: 슬라이딩 윈도우로 n_splits번 학습/검증 반복.
 
@@ -320,6 +362,9 @@ def walk_forward_auc(
     w = dataset["sample_weight"].values
     n = len(dataset)
 
+    lgbm_params, weight_scale = _load_lgbm_params(tuned_params_path)
+    w = (w * weight_scale).astype(np.float32)
+
     step = max(1, int(n * (1 - train_ratio) / n_splits))
     train_end_start = int(n * train_ratio)
 
@@ -340,18 +385,7 @@ def walk_forward_auc(
             neg_idx = np.random.choice(neg_idx, size=len(pos_idx), replace=False)
         idx = np.sort(np.concatenate([pos_idx, neg_idx]))
 
-        model = lgb.LGBMClassifier(
-            n_estimators=500,
-            learning_rate=0.05,
-            num_leaves=31,
-            min_child_samples=15,
-            subsample=0.8,
-            colsample_bytree=0.8,
-            reg_alpha=0.05,
-            reg_lambda=0.1,
-            random_state=42,
-            verbose=-1,
-        )
+        model = lgb.LGBMClassifier(**lgbm_params, random_state=42, verbose=-1)
         with warnings.catch_warnings():
             warnings.simplefilter("ignore")
             model.fit(X_tr[idx], y_tr[idx], sample_weight=w_tr[idx])
@@ -377,12 +411,21 @@ def main():
     )
     parser.add_argument("--wf", action="store_true", help="Walk-Forward 검증 실행")
     parser.add_argument("--wf-splits", type=int, default=5, help="Walk-Forward 폴드 수")
+    parser.add_argument(
+        "--tuned-params", type=str, default=None,
+        help="Optuna 튜닝 결과 JSON 경로 (지정 시 기본 파라미터를 덮어씀)",
+    )
     args = parser.parse_args()
 
     if args.wf:
-        walk_forward_auc(args.data, time_weight_decay=args.decay, n_splits=args.wf_splits)
+        walk_forward_auc(
+            args.data,
+            time_weight_decay=args.decay,
+            n_splits=args.wf_splits,
+            tuned_params_path=args.tuned_params,
+        )
     else:
-        train(args.data, time_weight_decay=args.decay)
+        train(args.data, time_weight_decay=args.decay, tuned_params_path=args.tuned_params)
 
 
 if __name__ == "__main__":

scripts/tune_hyperparams.py

@@ -0,0 +1,452 @@
+#!/usr/bin/env python3
+"""
+Optuna를 사용한 LightGBM 하이퍼파라미터 자동 탐색.
+
+사용법:
+    python scripts/tune_hyperparams.py                          # 기본 (50 trials, 5폴드)
+    python scripts/tune_hyperparams.py --trials 10 --folds 3   # 빠른 테스트
+    python scripts/tune_hyperparams.py --data data/combined_15m.parquet --trials 100
+    python scripts/tune_hyperparams.py --no-baseline            # 베이스라인 측정 건너뜀
+
+결과:
+    - 콘솔: Best Params + Walk-Forward 리포트
+    - JSON: models/tune_results_YYYYMMDD_HHMMSS.json
+"""
+import sys
+import warnings
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+import argparse
+import json
+import time
+from datetime import datetime
+
+import numpy as np
+import pandas as pd
+import lightgbm as lgb
+import optuna
+from optuna.samplers import TPESampler
+from optuna.pruners import MedianPruner
+from sklearn.metrics import roc_auc_score
+
+from src.ml_features import FEATURE_COLS
+from src.dataset_builder import generate_dataset_vectorized
+
+
+# ──────────────────────────────────────────────
+# 데이터 로드 및 데이터셋 생성 (1회 캐싱)
+# ──────────────────────────────────────────────
+
+def load_dataset(data_path: str) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
+    """
+    parquet 로드 → 벡터화 데이터셋 생성 → (X, y, w) numpy 배열 반환.
+    study 시작 전 1회만 호출하여 모든 trial이 공유한다.
+    """
+    print(f"데이터 로드: {data_path}")
+    df_raw = pd.read_parquet(data_path)
+    print(f"캔들 수: {len(df_raw):,}, 컬럼: {list(df_raw.columns)}")
+
+    base_cols = ["open", "high", "low", "close", "volume"]
+    btc_df = eth_df = None
+
+    if "close_btc" in df_raw.columns:
+        btc_df = df_raw[[c + "_btc" for c in base_cols]].copy()
+        btc_df.columns = base_cols
+        print("BTC 피처 활성화")
+
+    if "close_eth" in df_raw.columns:
+        eth_df = df_raw[[c + "_eth" for c in base_cols]].copy()
+        eth_df.columns = base_cols
+        print("ETH 피처 활성화")
+
+    df = df_raw[base_cols].copy()
+
+    print("\n데이터셋 생성 중 (1회만 실행)...")
+    dataset = generate_dataset_vectorized(df, btc_df=btc_df, eth_df=eth_df, time_weight_decay=0.0)
+
+    if dataset.empty or "label" not in dataset.columns:
+        raise ValueError("데이터셋 생성 실패: 샘플 0개")
+
+    actual_feature_cols = [c for c in FEATURE_COLS if c in dataset.columns]
+    X = dataset[actual_feature_cols].values.astype(np.float32)
+    y = dataset["label"].values.astype(np.int8)
+    w = dataset["sample_weight"].values.astype(np.float32)
+
+    pos = int(y.sum())
+    neg = int((y == 0).sum())
+    print(f"데이터셋 완성: {len(dataset):,}개 샘플 (양성={pos}, 음성={neg})")
+    print(f"사용 피처: {len(actual_feature_cols)}개\n")
+
+    return X, y, w
+
+
+# ──────────────────────────────────────────────
+# Walk-Forward 교차검증
+# ──────────────────────────────────────────────
+
+def _walk_forward_cv(
+    X: np.ndarray,
+    y: np.ndarray,
+    w: np.ndarray,
+    params: dict,
+    n_splits: int,
+    train_ratio: float,
+    trial: "optuna.Trial | None" = None,
+) -> tuple[float, list[float]]:
+    """
+    Walk-Forward 교차검증으로 평균 AUC를 반환한다.
+    trial이 제공되면 각 폴드 후 Optuna에 중간 값을 보고하여 Pruning을 활성화한다.
+    """
+    n = len(X)
+    step = max(1, int(n * (1 - train_ratio) / n_splits))
+    train_end_start = int(n * train_ratio)
+
+    fold_aucs: list[float] = []
+
+    for fold_idx in range(n_splits):
+        tr_end = train_end_start + fold_idx * step
+        val_end = tr_end + step
+        if val_end > n:
+            break
+
+        X_tr, y_tr, w_tr = X[:tr_end], y[:tr_end], w[:tr_end]
+        X_val, y_val = X[tr_end:val_end], y[tr_end:val_end]
+
+        # 클래스 불균형 처리: 언더샘플링 (시간 순서 유지)
+        pos_idx = np.where(y_tr == 1)[0]
+        neg_idx = np.where(y_tr == 0)[0]
+        if len(neg_idx) > len(pos_idx) and len(pos_idx) > 0:
+            rng = np.random.default_rng(42)
+            neg_idx = rng.choice(neg_idx, size=len(pos_idx), replace=False)
+        bal_idx = np.sort(np.concatenate([pos_idx, neg_idx]))
+
+        if len(bal_idx) < 20 or len(np.unique(y_val)) < 2:
+            fold_aucs.append(0.5)
+            continue
+
+        model = lgb.LGBMClassifier(**params, random_state=42, verbose=-1)
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            model.fit(X_tr[bal_idx], y_tr[bal_idx], sample_weight=w_tr[bal_idx])
+
+        proba = model.predict_proba(X_val)[:, 1]
+        auc = roc_auc_score(y_val, proba) if len(np.unique(y_val)) > 1 else 0.5
+        fold_aucs.append(float(auc))
+
+        # Optuna Pruning: 중간 값 보고
+        if trial is not None:
+            trial.report(float(np.mean(fold_aucs)), step=fold_idx)
+            if trial.should_prune():
+                raise optuna.TrialPruned()
+
+    mean_auc = float(np.mean(fold_aucs)) if fold_aucs else 0.5
+    return mean_auc, fold_aucs
+
+
+# ──────────────────────────────────────────────
+# Optuna 목적 함수
+# ──────────────────────────────────────────────
+
+def make_objective(
+    X: np.ndarray,
+    y: np.ndarray,
+    w: np.ndarray,
+    n_splits: int,
+    train_ratio: float,
+):
+    """클로저로 데이터셋을 캡처한 목적 함수를 반환한다."""
+
+    def objective(trial: optuna.Trial) -> float:
+        # ── 하이퍼파라미터 샘플링 ──
+        n_estimators     = trial.suggest_int("n_estimators", 100, 600)
+        learning_rate    = trial.suggest_float("learning_rate", 0.01, 0.2, log=True)
+        max_depth        = trial.suggest_int("max_depth", 2, 7)
+
+        # 핵심 제약: num_leaves <= 2^max_depth - 1 (leaf-wise 과적합 방지)
+        # 360개 수준의 소규모 데이터셋에서 num_leaves가 크면 암기 발생
+        max_leaves_upper = min(31, 2 ** max_depth - 1)
+        num_leaves       = trial.suggest_int("num_leaves", 7, max(7, max_leaves_upper))
+
+        min_child_samples = trial.suggest_int("min_child_samples", 10, 50)
+        subsample         = trial.suggest_float("subsample", 0.5, 1.0)
+        colsample_bytree  = trial.suggest_float("colsample_bytree", 0.5, 1.0)
+        reg_alpha         = trial.suggest_float("reg_alpha", 1e-4, 1.0, log=True)
+        reg_lambda        = trial.suggest_float("reg_lambda", 1e-4, 1.0, log=True)
+
+        # weight_scale: 데이터셋을 1회 캐싱하는 구조이므로
+        # time_weight_decay 효과를 sample_weight 스케일로 근사한다.
+        weight_scale = trial.suggest_float("weight_scale", 0.5, 2.0)
+        w_scaled = (w * weight_scale).astype(np.float32)
+
+        params = {
+            "n_estimators":     n_estimators,
+            "learning_rate":    learning_rate,
+            "max_depth":        max_depth,
+            "num_leaves":       num_leaves,
+            "min_child_samples": min_child_samples,
+            "subsample":        subsample,
+            "colsample_bytree": colsample_bytree,
+            "reg_alpha":        reg_alpha,
+            "reg_lambda":       reg_lambda,
+        }
+
+        mean_auc, fold_aucs = _walk_forward_cv(
+            X, y, w_scaled, params,
+            n_splits=n_splits,
+            train_ratio=train_ratio,
+            trial=trial,
+        )
+
+        # 폴드별 AUC를 user_attrs에 저장 (결과 리포트용)
+        trial.set_user_attr("fold_aucs", fold_aucs)
+
+        return mean_auc
+
+    return objective
+
+
+# ──────────────────────────────────────────────
+# 베이스라인 AUC 측정 (현재 고정 파라미터)
+# ──────────────────────────────────────────────
+
+def measure_baseline(
+    X: np.ndarray,
+    y: np.ndarray,
+    w: np.ndarray,
+    n_splits: int,
+    train_ratio: float,
+) -> tuple[float, list[float]]:
+    """현재 실전 파라미터(active 파일 또는 하드코딩 기본값)로 베이스라인 AUC를 측정한다."""
+    active_path = Path("models/active_lgbm_params.json")
+
+    if active_path.exists():
+        with open(active_path, "r", encoding="utf-8") as f:
+            tune_data = json.load(f)
+        best_params = dict(tune_data["best_trial"]["params"])
+        best_params.pop("weight_scale", None)
+        baseline_params = best_params
+        print(f"베이스라인 측정 중 (active 파일: {active_path})...")
+    else:
+        baseline_params = {
+            "n_estimators":      434,
+            "learning_rate":     0.123659,
+            "max_depth":         6,
+            "num_leaves":        14,
+            "min_child_samples": 10,
+            "subsample":         0.929062,
+            "colsample_bytree":  0.946330,
+            "reg_alpha":         0.573971,
+            "reg_lambda":        0.000157,
+        }
+        print("베이스라인 측정 중 (active 파일 없음 → 코드 내 기본 파라미터)...")
+
+    return _walk_forward_cv(X, y, w, baseline_params, n_splits=n_splits, train_ratio=train_ratio)
+
+
+# ──────────────────────────────────────────────
+# 결과 출력 및 저장
+# ──────────────────────────────────────────────
+
+def print_report(
+    study: optuna.Study,
+    baseline_auc: float,
+    baseline_folds: list[float],
+    elapsed_sec: float,
+    output_path: Path,
+) -> None:
+    """콘솔에 최종 리포트를 출력한다."""
+    best = study.best_trial
+    best_auc = best.value
+    best_folds = best.user_attrs.get("fold_aucs", [])
+    improvement = best_auc - baseline_auc
+    improvement_pct = (improvement / baseline_auc * 100) if baseline_auc > 0 else 0.0
+
+    elapsed_min = int(elapsed_sec // 60)
+    elapsed_s   = int(elapsed_sec % 60)
+
+    sep  = "=" * 64
+    dash = "-" * 64
+
+    completed = [t for t in study.trials if t.state == optuna.trial.TrialState.COMPLETE]
+    pruned    = [t for t in study.trials if t.state == optuna.trial.TrialState.PRUNED]
+
+    print(f"\n{sep}")
+    print(f"  Optuna 튜닝 완료 | {len(study.trials)} trials "
+          f"(완료={len(completed)}, 조기종료={len(pruned)}) | "
+          f"소요: {elapsed_min}분 {elapsed_s}초")
+    print(sep)
+    print(f"  Best AUC  : {best_auc:.4f}  (Trial #{best.number})")
+    if baseline_auc > 0:
+        sign = "+" if improvement >= 0 else ""
+        print(f"  Baseline  : {baseline_auc:.4f}  (현재 train_model.py 고정값)")
+        print(f"  개선폭    : {sign}{improvement:.4f} ({sign}{improvement_pct:.1f}%)")
+    print(dash)
+    print("  Best Parameters:")
+    for k, v in best.params.items():
+        if isinstance(v, float):
+            print(f"    {k:<22}: {v:.6f}")
+        else:
+            print(f"    {k:<22}: {v}")
+    print(dash)
+    print("  Walk-Forward 폴드별 AUC (Best Trial):")
+    for i, auc in enumerate(best_folds, 1):
+        print(f"    폴드 {i}: {auc:.4f}")
+    if best_folds:
+        arr = np.array(best_folds)
+        print(f"    평균: {arr.mean():.4f} ± {arr.std():.4f}")
+    if baseline_folds:
+        print(dash)
+        print("  Baseline 폴드별 AUC:")
+        for i, auc in enumerate(baseline_folds, 1):
+            print(f"    폴드 {i}: {auc:.4f}")
+        arr = np.array(baseline_folds)
+        print(f"    평균: {arr.mean():.4f} ± {arr.std():.4f}")
+    print(dash)
+    print(f"  결과 저장: {output_path}")
+    print(f"  다음 단계: python scripts/train_model.py  (파라미터 수동 반영 후)")
+    print(sep)
+
+
+def save_results(
+    study: optuna.Study,
+    baseline_auc: float,
+    baseline_folds: list[float],
+    elapsed_sec: float,
+    data_path: str,
+) -> Path:
+    """결과를 JSON 파일로 저장하고 경로를 반환한다."""
+    timestamp   = datetime.now().strftime("%Y%m%d_%H%M%S")
+    output_path = Path(f"models/tune_results_{timestamp}.json")
+    output_path.parent.mkdir(exist_ok=True)
+
+    best = study.best_trial
+
+    all_trials = []
+    for t in study.trials:
+        if t.state == optuna.trial.TrialState.COMPLETE:
+            all_trials.append({
+                "number":    t.number,
+                "auc":       round(t.value, 6),
+                "fold_aucs": [round(a, 6) for a in t.user_attrs.get("fold_aucs", [])],
+                "params":    {
+                    k: (round(v, 6) if isinstance(v, float) else v)
+                    for k, v in t.params.items()
+                },
+            })
+
+    result = {
+        "timestamp":        datetime.now().isoformat(),
+        "data_path":        data_path,
+        "n_trials_total":   len(study.trials),
+        "n_trials_complete": len(all_trials),
+        "elapsed_sec":      round(elapsed_sec, 1),
+        "baseline": {
+            "auc":       round(baseline_auc, 6),
+            "fold_aucs": [round(a, 6) for a in baseline_folds],
+        },
+        "best_trial": {
+            "number":    best.number,
+            "auc":       round(best.value, 6),
+            "fold_aucs": [round(a, 6) for a in best.user_attrs.get("fold_aucs", [])],
+            "params":    {
+                k: (round(v, 6) if isinstance(v, float) else v)
+                for k, v in best.params.items()
+            },
+        },
+        "all_trials": all_trials,
+    }
+
+    with open(output_path, "w", encoding="utf-8") as f:
+        json.dump(result, f, indent=2, ensure_ascii=False)
+
+    return output_path
+
+
+# ──────────────────────────────────────────────
+# 메인
+# ──────────────────────────────────────────────
+
+def main():
+    parser = argparse.ArgumentParser(description="Optuna LightGBM 하이퍼파라미터 튜닝")
+    parser.add_argument("--data",        default="data/combined_15m.parquet", help="학습 데이터 경로")
+    parser.add_argument("--trials",      type=int,   default=50,  help="Optuna trial 수 (기본: 50)")
+    parser.add_argument("--folds",       type=int,   default=5,   help="Walk-Forward 폴드 수 (기본: 5)")
+    parser.add_argument("--train-ratio", type=float, default=0.6, help="학습 구간 비율 (기본: 0.6)")
+    parser.add_argument("--no-baseline", action="store_true",     help="베이스라인 측정 건너뜀")
+    args = parser.parse_args()
+
+    # 1. 데이터셋 로드 (1회)
+    X, y, w = load_dataset(args.data)
+
+    # 2. 베이스라인 측정
+    if args.no_baseline:
+        baseline_auc, baseline_folds = 0.0, []
+        print("베이스라인 측정 건너뜀 (--no-baseline)\n")
+    else:
+        baseline_auc, baseline_folds = measure_baseline(X, y, w, args.folds, args.train_ratio)
+        print(
+            f"베이스라인 AUC: {baseline_auc:.4f} "
+            f"(폴드별: {[round(a, 4) for a in baseline_folds]})\n"
+        )
+
+    # 3. Optuna study 실행
+    optuna.logging.set_verbosity(optuna.logging.WARNING)
+    sampler = TPESampler(seed=42)
+    pruner  = MedianPruner(n_startup_trials=5, n_warmup_steps=2)
+    study   = optuna.create_study(
+        direction="maximize",
+        sampler=sampler,
+        pruner=pruner,
+        study_name="lgbm_wf_auc",
+    )
+
+    objective = make_objective(X, y, w, n_splits=args.folds, train_ratio=args.train_ratio)
+
+    print(f"Optuna 탐색 시작: {args.trials} trials, {args.folds}폴드 Walk-Forward")
+    print("(trial 완료마다 진행 상황 출력)\n")
+
+    start_time = time.time()
+
+    def _progress_callback(study: optuna.Study, trial: optuna.trial.FrozenTrial) -> None:
+        if trial.state == optuna.trial.TrialState.COMPLETE:
+            best_so_far = study.best_value
+            leaves  = trial.params.get("num_leaves", "?")
+            depth   = trial.params.get("max_depth", "?")
+            print(
+                f"  Trial #{trial.number:3d} | AUC={trial.value:.4f} "
+                f"| Best={best_so_far:.4f} "
+                f"| leaves={leaves} depth={depth}"
+            )
+        elif trial.state == optuna.trial.TrialState.PRUNED:
+            print(f"  Trial #{trial.number:3d} | PRUNED (조기 종료)")
+
+    study.optimize(
+        objective,
+        n_trials=args.trials,
+        callbacks=[_progress_callback],
+        show_progress_bar=False,
+    )
+
+    elapsed = time.time() - start_time
+
+    # 4. 결과 저장 및 출력
+    output_path = save_results(study, baseline_auc, baseline_folds, elapsed, args.data)
+    print_report(study, baseline_auc, baseline_folds, elapsed, output_path)
+
+    # 5. 성능 개선 시 active 파일 자동 갱신
+    import shutil
+    active_path = Path("models/active_lgbm_params.json")
+    if not args.no_baseline and study.best_value > baseline_auc:
+        shutil.copy(output_path, active_path)
+        improvement = study.best_value - baseline_auc
+        print(f"[MLOps] AUC +{improvement:.4f} 개선 → {active_path} 자동 갱신 완료")
+        print(f"[MLOps] 다음 train_model.py 실행 시 새 파라미터가 자동 적용됩니다.\n")
+    elif args.no_baseline:
+        print("[MLOps] --no-baseline 모드: 성능 비교 없이 active 파일 유지\n")
+    else:
+        print(f"[MLOps] 성능 개선 없음 (Best={study.best_value:.4f} ≤ Baseline={baseline_auc:.4f}) → active 파일 유지\n")
+
+
+if __name__ == "__main__":
+    main()

src/bot.py

@@ -1,4 +1,5 @@
 import asyncio
+import pandas as pd
 from loguru import logger
 from src.config import Config
 from src.exchange import BinanceFuturesClient
@@ -8,6 +9,7 @@ from src.notifier import DiscordNotifier
 from src.risk_manager import RiskManager
 from src.ml_filter import MLFilter
 from src.ml_features import build_features
+from src.user_data_stream import UserDataStream
 
 
 class TradingBot:
@@ -18,18 +20,22 @@ class TradingBot:
         self.risk = RiskManager(config)
         self.ml_filter = MLFilter()
         self.current_trade_side: str | None = None  # "LONG" | "SHORT"
+        self._entry_price: float | None = None
+        self._entry_quantity: float | None = None
+        self._is_reentering: bool = False  # _close_and_reenter 중 콜백 상태 초기화 방지
+        self._prev_oi: float | None = None  # OI 변화율 계산용 이전 값
         self.stream = MultiSymbolStream(
             symbols=[config.symbol, "BTCUSDT", "ETHUSDT"],
             interval="15m",
             on_candle=self._on_candle_closed,
         )
 
-    def _on_candle_closed(self, candle: dict):
+    async def _on_candle_closed(self, candle: dict):
         xrp_df = self.stream.get_dataframe(self.config.symbol)
         btc_df = self.stream.get_dataframe("BTCUSDT")
         eth_df = self.stream.get_dataframe("ETHUSDT")
         if xrp_df is not None:
-            asyncio.create_task(self.process_candle(xrp_df, btc_df=btc_df, eth_df=eth_df))
+            await self.process_candle(xrp_df, btc_df=btc_df, eth_df=eth_df)
 
     async def _recover_position(self) -> None:
         """재시작 시 바이낸스에서 현재 포지션을 조회하여 상태 복구."""
@@ -37,6 +43,8 @@ class TradingBot:
         if position is not None:
             amt = float(position["positionAmt"])
             self.current_trade_side = "LONG" if amt > 0 else "SHORT"
+            self._entry_price = float(position["entryPrice"])
+            self._entry_quantity = abs(amt)
             entry = float(position["entryPrice"])
             logger.info(
                 f"기존 포지션 복구: {self.current_trade_side} | "
@@ -49,40 +57,76 @@ class TradingBot:
         else:
             logger.info("기존 포지션 없음 - 신규 진입 대기")
 
+    async def _fetch_market_microstructure(self) -> tuple[float, float]:
+        """OI 변화율과 펀딩비를 실시간으로 조회한다. 실패 시 0.0으로 폴백."""
+        oi_val, fr_val = await asyncio.gather(
+            self.exchange.get_open_interest(),
+            self.exchange.get_funding_rate(),
+            return_exceptions=True,
+        )
+        # None(API 실패) 또는 Exception이면 _calc_oi_change를 호출하지 않고 0.0 반환
+        if isinstance(oi_val, (int, float)) and oi_val > 0:
+            oi_change = self._calc_oi_change(float(oi_val))
+        else:
+            oi_change = 0.0
+        fr_float = float(fr_val) if isinstance(fr_val, (int, float)) else 0.0
+        logger.debug(f"OI={oi_val}, OI변화율={oi_change:.6f}, 펀딩비={fr_float:.6f}")
+        return oi_change, fr_float
+
+    def _calc_oi_change(self, current_oi: float) -> float:
+        """이전 OI 대비 변화율을 계산한다. 첫 캔들은 0.0 반환."""
+        if self._prev_oi is None or self._prev_oi == 0.0:
+            self._prev_oi = current_oi
+            return 0.0
+        change = (current_oi - self._prev_oi) / self._prev_oi
+        self._prev_oi = current_oi
+        return change
+
     async def process_candle(self, df, btc_df=None, eth_df=None):
         self.ml_filter.check_and_reload()
 
+        # 캔들 마감 시 OI/펀딩비 실시간 조회 (실패해도 0으로 폴백)
+        oi_change, funding_rate = await self._fetch_market_microstructure()
+
         if not self.risk.is_trading_allowed():
             logger.warning("리스크 한도 초과 - 거래 중단")
             return
 
         ind = Indicators(df)
         df_with_indicators = ind.calculate_all()
-        signal = ind.get_signal(df_with_indicators)
-
-        if signal != "HOLD" and self.ml_filter.is_model_loaded():
-            features = build_features(df_with_indicators, signal, btc_df=btc_df, eth_df=eth_df)
-            if not self.ml_filter.should_enter(features):
-                logger.info(f"ML 필터 차단: {signal} 신호 무시")
-                signal = "HOLD"
+        raw_signal = ind.get_signal(df_with_indicators)
 
         current_price = df_with_indicators["close"].iloc[-1]
-        logger.info(f"신호: {signal} | 현재가: {current_price:.4f} USDT")
+        logger.info(f"신호: {raw_signal} | 현재가: {current_price:.4f} USDT")
 
         position = await self.exchange.get_position()
 
-        if position is None and signal != "HOLD":
+        if position is None and raw_signal != "HOLD":
             self.current_trade_side = None
             if not self.risk.can_open_new_position():
                 logger.info("최대 포지션 수 도달")
                 return
+            signal = raw_signal
+            features = build_features(
+                df_with_indicators, signal,
+                btc_df=btc_df, eth_df=eth_df,
+                oi_change=oi_change, funding_rate=funding_rate,
+            )
+            if self.ml_filter.is_model_loaded():
+                if not self.ml_filter.should_enter(features):
+                    logger.info(f"ML 필터 차단: {signal} 신호 무시")
+                    return
             await self._open_position(signal, df_with_indicators)
 
         elif position is not None:
             pos_side = "LONG" if float(position["positionAmt"]) > 0 else "SHORT"
-            if (pos_side == "LONG" and signal == "SHORT") or \
-               (pos_side == "SHORT" and signal == "LONG"):
-                await self._close_position(position)
+            if (pos_side == "LONG" and raw_signal == "SHORT") or \
+               (pos_side == "SHORT" and raw_signal == "LONG"):
+                await self._close_and_reenter(
+                    position, raw_signal, df_with_indicators,
+                    btc_df=btc_df, eth_df=eth_df,
+                    oi_change=oi_change, funding_rate=funding_rate,
+                )
 
     async def _open_position(self, signal: str, df):
         balance = await self.exchange.get_balance()
@@ -108,12 +152,14 @@ class TradingBot:
 
         last_row = df.iloc[-1]
         signal_snapshot = {
-            "rsi":       float(last_row.get("rsi", 0)),
-            "macd_hist": float(last_row.get("macd_hist", 0)),
-            "atr":       float(last_row.get("atr", 0)),
+            "rsi":       float(last_row["rsi"])       if "rsi"       in last_row.index and pd.notna(last_row["rsi"])       else 0.0,
+            "macd_hist": float(last_row["macd_hist"]) if "macd_hist" in last_row.index and pd.notna(last_row["macd_hist"]) else 0.0,
+            "atr":       float(last_row["atr"])       if "atr"       in last_row.index and pd.notna(last_row["atr"])       else 0.0,
         }
 
         self.current_trade_side = signal
+        self._entry_price = price
+        self._entry_quantity = quantity
         self.notifier.notify_open(
             symbol=self.config.symbol,
             side=signal,
@@ -145,26 +191,91 @@ class TradingBot:
             reduce_only=True,
         )
 
-    async def _close_position(self, position: dict):
-        amt = abs(float(position["positionAmt"]))
-        side = "SELL" if float(position["positionAmt"]) > 0 else "BUY"
-        pos_side = "LONG" if side == "SELL" else "SHORT"
-        await self.exchange.cancel_all_orders()
-        await self.exchange.place_order(side=side, quantity=amt, reduce_only=True)
+    def _calc_estimated_pnl(self, exit_price: float) -> float:
+        """진입가·수량 기반 예상 PnL 계산 (수수료 미반영)."""
+        if self._entry_price is None or self._entry_quantity is None or self.current_trade_side is None:
+            return 0.0
+        if self.current_trade_side == "LONG":
+            return (exit_price - self._entry_price) * self._entry_quantity
+        return (self._entry_price - exit_price) * self._entry_quantity
 
-        entry = float(position["entryPrice"])
-        mark  = float(position["markPrice"])
-        pnl   = (mark - entry) * amt if side == "SELL" else (entry - mark) * amt
+    async def _on_position_closed(
+        self,
+        net_pnl: float,
+        close_reason: str,
+        exit_price: float,
+    ) -> None:
+        """User Data Stream에서 청산 감지 시 호출되는 콜백."""
+        estimated_pnl = self._calc_estimated_pnl(exit_price)
+        diff = net_pnl - estimated_pnl
+
+        self.risk.record_pnl(net_pnl)
 
         self.notifier.notify_close(
             symbol=self.config.symbol,
-            side=pos_side,
-            exit_price=mark,
-            pnl=pnl,
+            side=self.current_trade_side or "UNKNOWN",
+            close_reason=close_reason,
+            exit_price=exit_price,
+            estimated_pnl=estimated_pnl,
+            net_pnl=net_pnl,
+            diff=diff,
         )
-        self.risk.record_pnl(pnl)
+
+        logger.success(
+            f"포지션 청산({close_reason}): 예상={estimated_pnl:+.4f}, "
+            f"순수익={net_pnl:+.4f}, 차이={diff:+.4f} USDT"
+        )
+
+        # _close_and_reenter 중이면 신규 포지션 상태를 덮어쓰지 않는다
+        if self._is_reentering:
+            return
+
+        # Flat 상태로 초기화
         self.current_trade_side = None
-        logger.success(f"포지션 청산: PnL={pnl:.4f} USDT")
+        self._entry_price = None
+        self._entry_quantity = None
+
+    async def _close_position(self, position: dict):
+        """포지션 청산 주문만 실행한다. PnL 기록/알림은 _on_position_closed 콜백이 담당."""
+        amt = abs(float(position["positionAmt"]))
+        side = "SELL" if float(position["positionAmt"]) > 0 else "BUY"
+        await self.exchange.cancel_all_orders()
+        await self.exchange.place_order(side=side, quantity=amt, reduce_only=True)
+        logger.info(f"청산 주문 전송 완료 (side={side}, qty={amt})")
+
+    async def _close_and_reenter(
+        self,
+        position: dict,
+        signal: str,
+        df,
+        btc_df=None,
+        eth_df=None,
+        oi_change: float = 0.0,
+        funding_rate: float = 0.0,
+    ) -> None:
+        """기존 포지션을 청산하고, ML 필터 통과 시 반대 방향으로 즉시 재진입한다."""
+        # 재진입 플래그: User Data Stream 콜백이 신규 포지션 상태를 초기화하지 않도록 보호
+        self._is_reentering = True
+        try:
+            await self._close_position(position)
+
+            if not self.risk.can_open_new_position():
+                logger.info("최대 포지션 수 도달 — 재진입 건너뜀")
+                return
+
+            if self.ml_filter.is_model_loaded():
+                features = build_features(
+                    df, signal,
+                    btc_df=btc_df, eth_df=eth_df,
+                    oi_change=oi_change, funding_rate=funding_rate,
+                )
+                if not self.ml_filter.should_enter(features):
+                    logger.info(f"ML 필터 차단: {signal} 재진입 무시")
+                    return
+
+            await self._open_position(signal, df)
+        finally:
+            self._is_reentering = False
 
     async def run(self):
         logger.info(f"봇 시작: {self.config.symbol}, 레버리지 {self.config.leverage}x")
@@ -172,7 +283,19 @@ class TradingBot:
         balance = await self.exchange.get_balance()
         self.risk.set_base_balance(balance)
         logger.info(f"기준 잔고 설정: {balance:.2f} USDT (동적 증거금 비율 기준점)")
-        await self.stream.start(
-            api_key=self.config.api_key,
-            api_secret=self.config.api_secret,
+
+        user_stream = UserDataStream(
+            symbol=self.config.symbol,
+            on_order_filled=self._on_position_closed,
+        )
+
+        await asyncio.gather(
+            self.stream.start(
+                api_key=self.config.api_key,
+                api_secret=self.config.api_secret,
+            ),
+            user_stream.start(
+                api_key=self.config.api_key,
+                api_secret=self.config.api_secret,
+            ),
         )

src/data_stream.py

@@ -40,12 +40,12 @@ class KlineStream:
             "is_closed": k["x"],
         }
 
-    def handle_message(self, msg: dict):
+    async def handle_message(self, msg: dict):
         candle = self.parse_kline(msg)
         if candle["is_closed"]:
             self.buffer.append(candle)
             if self.on_candle:
-                self.on_candle(candle)
+                await self.on_candle(candle)
 
     def get_dataframe(self) -> pd.DataFrame | None:
         if len(self.buffer) < _MIN_CANDLES_FOR_SIGNAL:
@@ -90,7 +90,7 @@ class KlineStream:
             ) as stream:
                 while True:
                     msg = await stream.recv()
-                    self.handle_message(msg)
+                    await self.handle_message(msg)
         finally:
             await client.close_connection()
 
@@ -129,7 +129,7 @@ class MultiSymbolStream:
             "is_closed": k["x"],
         }
 
-    def handle_message(self, msg: dict):
+    async def handle_message(self, msg: dict):
         # Combined stream 메시지는 {"stream": "...", "data": {...}} 형태
         if "stream" in msg:
             data = msg["data"]
@@ -145,7 +145,7 @@ class MultiSymbolStream:
         if candle["is_closed"] and symbol in self.buffers:
             self.buffers[symbol].append(candle)
             if symbol == self.primary_symbol and self.on_candle:
-                self.on_candle(candle)
+                await self.on_candle(candle)
 
     def get_dataframe(self, symbol: str) -> pd.DataFrame | None:
         key = symbol.lower()
@@ -192,6 +192,6 @@ class MultiSymbolStream:
             async with bm.futures_multiplex_socket(streams) as stream:
                 while True:
                     msg = await stream.recv()
-                    self.handle_message(msg)
+                    await self.handle_message(msg)
         finally:
             await client.close_connection()

src/dataset_builder.py

@@ -242,8 +242,16 @@ def _calc_features_vectorized(
         eth_r5 = _align(eth_ret_5, n).astype(np.float32)
 
         xrp_r1 = ret_1.astype(np.float32)
-        xrp_btc_rs_raw = (xrp_r1 / (btc_r1 + 1e-8)).astype(np.float32)
-        xrp_eth_rs_raw = (xrp_r1 / (eth_r1 + 1e-8)).astype(np.float32)
+        xrp_btc_rs_raw = np.divide(
+            xrp_r1, btc_r1,
+            out=np.zeros_like(xrp_r1),
+            where=(btc_r1 != 0),
+        ).astype(np.float32)
+        xrp_eth_rs_raw = np.divide(
+            xrp_r1, eth_r1,
+            out=np.zeros_like(xrp_r1),
+            where=(eth_r1 != 0),
+        ).astype(np.float32)
 
         extra = pd.DataFrame({
             "btc_ret_1":  _rolling_zscore(btc_r1),

src/exchange.py

@@ -45,6 +45,8 @@ class BinanceFuturesClient:
                 return float(b["balance"])
         return 0.0
 
+    _ALGO_ORDER_TYPES = {"STOP_MARKET", "TAKE_PROFIT_MARKET", "STOP", "TAKE_PROFIT", "TRAILING_STOP_MARKET"}
+
     async def place_order(
         self,
         side: str,
@@ -55,6 +57,16 @@ class BinanceFuturesClient:
         reduce_only: bool = False,
     ) -> dict:
         loop = asyncio.get_event_loop()
+
+        if order_type in self._ALGO_ORDER_TYPES:
+            return await self._place_algo_order(
+                side=side,
+                quantity=quantity,
+                order_type=order_type,
+                stop_price=stop_price,
+                reduce_only=reduce_only,
+            )
+
         params = dict(
             symbol=self.config.symbol,
             side=side,
@@ -75,6 +87,34 @@ class BinanceFuturesClient:
             logger.error(f"주문 실패: {e}")
             raise
 
+    async def _place_algo_order(
+        self,
+        side: str,
+        quantity: float,
+        order_type: str,
+        stop_price: float = None,
+        reduce_only: bool = False,
+    ) -> dict:
+        """STOP_MARKET / TAKE_PROFIT_MARKET 등 Algo Order API(/fapi/v1/algoOrder)로 전송."""
+        loop = asyncio.get_event_loop()
+        params = dict(
+            symbol=self.config.symbol,
+            side=side,
+            algoType="CONDITIONAL",
+            type=order_type,
+            quantity=quantity,
+            reduceOnly="true" if reduce_only else "false",
+        )
+        if stop_price:
+            params["triggerPrice"] = stop_price
+        try:
+            return await loop.run_in_executor(
+                None, lambda: self.client.futures_create_algo_order(**params)
+            )
+        except BinanceAPIException as e:
+            logger.error(f"Algo 주문 실패: {e}")
+            raise
+
     async def get_position(self) -> dict | None:
         loop = asyncio.get_event_loop()
         positions = await loop.run_in_executor(
@@ -89,10 +129,74 @@ class BinanceFuturesClient:
         return None
 
     async def cancel_all_orders(self):
+        """일반 오픈 주문과 Algo 오픈 주문을 모두 취소한다."""
         loop = asyncio.get_event_loop()
-        return await loop.run_in_executor(
+        await loop.run_in_executor(
             None,
             lambda: self.client.futures_cancel_all_open_orders(
                 symbol=self.config.symbol
             ),
         )
+        try:
+            await loop.run_in_executor(
+                None,
+                lambda: self.client.futures_cancel_all_algo_open_orders(
+                    symbol=self.config.symbol
+                ),
+            )
+        except Exception as e:
+            logger.warning(f"Algo 주문 전체 취소 실패 (무시): {e}")
+
+    async def get_open_interest(self) -> float | None:
+        """현재 미결제약정(OI)을 조회한다. 오류 시 None 반환."""
+        loop = asyncio.get_event_loop()
+        try:
+            result = await loop.run_in_executor(
+                None,
+                lambda: self.client.futures_open_interest(symbol=self.config.symbol),
+            )
+            return float(result["openInterest"])
+        except Exception as e:
+            logger.warning(f"OI 조회 실패 (무시): {e}")
+            return None
+
+    async def get_funding_rate(self) -> float | None:
+        """현재 펀딩비를 조회한다. 오류 시 None 반환."""
+        loop = asyncio.get_event_loop()
+        try:
+            result = await loop.run_in_executor(
+                None,
+                lambda: self.client.futures_mark_price(symbol=self.config.symbol),
+            )
+            return float(result["lastFundingRate"])
+        except Exception as e:
+            logger.warning(f"펀딩비 조회 실패 (무시): {e}")
+            return None
+
+    async def create_listen_key(self) -> str:
+        """POST /fapi/v1/listenKey — listenKey 신규 발급"""
+        loop = asyncio.get_event_loop()
+        result = await loop.run_in_executor(
+            None,
+            lambda: self.client.futures_stream_get_listen_key(),
+        )
+        return result
+
+    async def keepalive_listen_key(self, listen_key: str) -> None:
+        """PUT /fapi/v1/listenKey — listenKey 만료 연장 (60분 → 리셋)"""
+        loop = asyncio.get_event_loop()
+        await loop.run_in_executor(
+            None,
+            lambda: self.client.futures_stream_keepalive(listenKey=listen_key),
+        )
+
+    async def delete_listen_key(self, listen_key: str) -> None:
+        """DELETE /fapi/v1/listenKey — listenKey 삭제 (정상 종료 시)"""
+        loop = asyncio.get_event_loop()
+        try:
+            await loop.run_in_executor(
+                None,
+                lambda: self.client.futures_stream_close(listenKey=listen_key),
+            )
+        except Exception as e:
+            logger.warning(f"listenKey 삭제 실패 (무시): {e}")

src/ml_features.py

@@ -34,11 +34,14 @@ def build_features(
     signal: str,
     btc_df: pd.DataFrame | None = None,
     eth_df: pd.DataFrame | None = None,
+    oi_change: float | None = None,
+    funding_rate: float | None = None,
 ) -> pd.Series:
     """
     기술 지표가 계산된 DataFrame의 마지막 행에서 ML 피처를 추출한다.
-    btc_df, eth_df가 제공되면 21개 피처를, 없으면 13개 피처를 반환한다.
+    btc_df, eth_df가 제공되면 23개 피처를, 없으면 15개 피처를 반환한다.
     signal: "LONG" | "SHORT"
+    oi_change, funding_rate: 실제 값이 제공되면 사용, 없으면 0.0으로 채운다.
     """
     last = df.iloc[-1]
     close = last["close"]
@@ -127,4 +130,8 @@ def build_features(
             "xrp_eth_rs": float(_calc_rs(ret_1, eth_ret_1)),
         })
 
+    # 실시간에서 실제 값이 제공되면 사용, 없으면 0으로 채운다
+    base["oi_change"]    = float(oi_change)    if oi_change    is not None else 0.0
+    base["funding_rate"] = float(funding_rate) if funding_rate is not None else 0.0
+
     return pd.Series(base)

src/ml_filter.py

@@ -53,8 +53,12 @@ class MLFilter:
         if self._onnx_path.exists():
             try:
                 import onnxruntime as ort
+                sess_opts = ort.SessionOptions()
+                sess_opts.intra_op_num_threads = 1
+                sess_opts.inter_op_num_threads = 1
                 self._onnx_session = ort.InferenceSession(
                     str(self._onnx_path),
+                    sess_options=sess_opts,
                     providers=["CPUExecutionProvider"],
                 )
                 self._lgbm_model = None

src/mlx_filter.py

@@ -71,15 +71,20 @@ def _export_onnx(
                          transB=1),
         # sigmoid → (N, 1)
         helper.make_node("Sigmoid", ["logits"],          ["proba_2d"]),
-        # squeeze: (N, 1) → (N,)
-        helper.make_node("Flatten", ["proba_2d"],        ["proba"], axis=0),
+        # squeeze: (N, 1) → (N,)  — axis=-1 로 마지막 차원만 제거
+        helper.make_node("Squeeze", ["proba_2d", "squeeze_axes"], ["proba"]),
     ]
 
+    squeeze_axes = numpy_helper.from_array(
+        np.array([-1], dtype=np.int64), name="squeeze_axes"
+    )
+    initializers.append(squeeze_axes)
+
     graph = helper.make_graph(
         nodes,
         "mlx_filter",
         inputs=[helper.make_tensor_value_info("X", TensorProto.FLOAT, [None, input_dim])],
-        outputs=[helper.make_tensor_value_info("proba", TensorProto.FLOAT, [None])],
+        outputs=[helper.make_tensor_value_info("proba", TensorProto.FLOAT, [-1])],
         initializer=initializers,
     )
     model_proto = helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)])
@@ -142,8 +147,10 @@ class MLXFilter:
 
         # nan-safe 정규화: nanmean/nanstd로 통계 계산 후 nan → 0.0 대치
         # (z-score 후 0.0 = 평균값, 신경망에 줄 수 있는 가장 무난한 결측 대치값)
-        self._mean = np.nanmean(X_np, axis=0)
-        self._std  = np.nanstd(X_np, axis=0) + 1e-8
+        mean_vals  = np.nanmean(X_np, axis=0)
+        self._mean = np.nan_to_num(mean_vals, nan=0.0)   # 전체-NaN 컬럼 → 평균 0.0
+        std_vals   = np.nanstd(X_np, axis=0)
+        self._std  = np.nan_to_num(std_vals, nan=1.0) + 1e-8  # 전체-NaN 컬럼 → std 1.0
         X_np = (X_np - self._mean) / self._std
         X_np = np.nan_to_num(X_np, nan=0.0)
 

src/notifier.py

@@ -49,13 +49,20 @@ class DiscordNotifier:
         self,
         symbol: str,
         side: str,
+        close_reason: str,
         exit_price: float,
-        pnl: float,
+        estimated_pnl: float,
+        net_pnl: float,
+        diff: float,
     ) -> None:
-        emoji = "✅" if pnl >= 0 else "❌"
+        emoji_map = {"TP": "✅", "SL": "❌", "MANUAL": "🔶"}
+        emoji = emoji_map.get(close_reason, "🔶")
         msg = (
-            f"{emoji} **[{symbol}] {side} 청산**\n"
-            f"청산가: `{exit_price:.4f}` | PnL: `{pnl:+.4f} USDT`"
+            f"{emoji} **[{symbol}] {side} {close_reason} 청산**\n"
+            f"청산가:               `{exit_price:.4f}`\n"
+            f"예상 수익:            `{estimated_pnl:+.4f} USDT`\n"
+            f"실제 순수익:          `{net_pnl:+.4f} USDT`\n"
+            f"차이(슬리피지+수수료): `{diff:+.4f} USDT`"
         )
         self._send(msg)
 

src/user_data_stream.py

@@ -0,0 +1,105 @@
+import asyncio
+from typing import Callable
+from binance import AsyncClient, BinanceSocketManager
+from loguru import logger
+
+_RECONNECT_DELAY = 5  # 재연결 대기 초
+
+_CLOSE_ORDER_TYPES = {"TAKE_PROFIT_MARKET", "STOP_MARKET"}
+
+
+class UserDataStream:
+    """
+    Binance Futures User Data Stream을 구독하여 주문 체결 이벤트를 처리한다.
+
+    - python-binance BinanceSocketManager의 내장 keepalive 활용
+    - 네트워크 단절 시 무한 재연결 루프
+    - ORDER_TRADE_UPDATE 이벤트에서 지정 심볼의 청산 주문만 필터링하여 콜백 호출
+    """
+
+    def __init__(
+        self,
+        symbol: str,                     # 감시할 심볼 (예: "XRPUSDT")
+        on_order_filled: Callable,       # bot._on_position_closed 콜백
+    ):
+        self._symbol = symbol.upper()
+        self._on_order_filled = on_order_filled
+
+    async def start(self, api_key: str, api_secret: str) -> None:
+        """User Data Stream 메인 루프 — 봇 종료 시까지 실행."""
+        client = await AsyncClient.create(
+            api_key=api_key,
+            api_secret=api_secret,
+        )
+        bm = BinanceSocketManager(client)
+        try:
+            await self._run_loop(bm)
+        finally:
+            await client.close_connection()
+
+    async def _run_loop(self, bm: BinanceSocketManager) -> None:
+        """연결 → 재연결 무한 루프. BinanceSocketManager가 listenKey keepalive를 내부 처리한다."""
+        while True:
+            try:
+                async with bm.futures_user_socket() as stream:
+                    logger.info(f"User Data Stream 연결 완료 (심볼 필터: {self._symbol})")
+                    async for msg in stream:
+                        await self._handle_message(msg)
+
+            except asyncio.CancelledError:
+                logger.info("User Data Stream 정상 종료")
+                raise
+
+            except Exception as e:
+                logger.warning(
+                    f"User Data Stream 끊김: {e} — "
+                    f"{_RECONNECT_DELAY}초 후 재연결"
+                )
+                await asyncio.sleep(_RECONNECT_DELAY)
+
+    async def _handle_message(self, msg: dict) -> None:
+        """ORDER_TRADE_UPDATE 이벤트에서 청산 주문을 필터링하여 콜백을 호출한다."""
+        if msg.get("e") != "ORDER_TRADE_UPDATE":
+            return
+
+        order = msg.get("o", {})
+
+        # 심볼 필터링: 봇이 관리하는 심볼만 처리
+        if order.get("s", "") != self._symbol:
+            return
+
+        # x: Execution Type, X: Order Status
+        if order.get("x") != "TRADE" or order.get("X") != "FILLED":
+            return
+
+        order_type   = order.get("o", "")
+        is_reduce    = order.get("R", False)
+        realized_pnl = float(order.get("rp", "0"))
+
+        # 청산 주문 판별: reduceOnly이거나, TP/SL 타입이거나, rp != 0
+        is_close = is_reduce or order_type in _CLOSE_ORDER_TYPES or realized_pnl != 0
+        if not is_close:
+            return
+
+        commission = abs(float(order.get("n", "0")))
+        net_pnl    = realized_pnl - commission
+        exit_price = float(order.get("ap", "0"))
+
+        if order_type == "TAKE_PROFIT_MARKET":
+            close_reason = "TP"
+        elif order_type == "STOP_MARKET":
+            close_reason = "SL"
+        else:
+            close_reason = "MANUAL"
+
+        logger.info(
+            f"청산 감지({close_reason}): exit={exit_price:.4f}, "
+            f"rp={realized_pnl:+.4f}, commission={commission:.4f}, "
+            f"net_pnl={net_pnl:+.4f}"
+        )
+
+        await self._on_order_filled(
+            net_pnl=net_pnl,
+            close_reason=close_reason,
+            exit_price=exit_price,
+        )

tests/test_bot.py

@@ -69,3 +69,179 @@ async def test_bot_processes_signal(config, sample_df):
         mock_ind.get_atr_stop.return_value = (0.48, 0.56)
         MockInd.return_value = mock_ind
         await bot.process_candle(sample_df)
+
+
+@pytest.mark.asyncio
+async def test_close_and_reenter_calls_open_when_ml_passes(config, sample_df):
+    """반대 시그널 + ML 필터 통과 시 청산 후 재진입해야 한다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot._close_position = AsyncMock()
+    bot._open_position = AsyncMock()
+    bot.risk = MagicMock()
+    bot.risk.can_open_new_position.return_value = True
+    bot.ml_filter = MagicMock()
+    bot.ml_filter.is_model_loaded.return_value = True
+    bot.ml_filter.should_enter.return_value = True
+
+    position = {"positionAmt": "100", "entryPrice": "0.5", "markPrice": "0.52"}
+    await bot._close_and_reenter(position, "SHORT", sample_df)
+
+    bot._close_position.assert_awaited_once_with(position)
+    bot._open_position.assert_awaited_once_with("SHORT", sample_df)
+
+
+@pytest.mark.asyncio
+async def test_close_and_reenter_skips_open_when_ml_blocks(config, sample_df):
+    """ML 필터 차단 시 청산만 하고 재진입하지 않아야 한다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot._close_position = AsyncMock()
+    bot._open_position = AsyncMock()
+    bot.ml_filter = MagicMock()
+    bot.ml_filter.is_model_loaded.return_value = True
+    bot.ml_filter.should_enter.return_value = False
+
+    position = {"positionAmt": "100", "entryPrice": "0.5", "markPrice": "0.52"}
+    await bot._close_and_reenter(position, "SHORT", sample_df)
+
+    bot._close_position.assert_awaited_once_with(position)
+    bot._open_position.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_close_and_reenter_skips_open_when_max_positions_reached(config, sample_df):
+    """최대 포지션 수 도달 시 청산만 하고 재진입하지 않아야 한다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot._close_position = AsyncMock()
+    bot._open_position = AsyncMock()
+    bot.risk = MagicMock()
+    bot.risk.can_open_new_position.return_value = False
+
+    position = {"positionAmt": "100", "entryPrice": "0.5", "markPrice": "0.52"}
+    await bot._close_and_reenter(position, "SHORT", sample_df)
+
+    bot._close_position.assert_awaited_once_with(position)
+    bot._open_position.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_process_candle_calls_close_and_reenter_on_reverse_signal(config, sample_df):
+    """반대 시그널 시 process_candle이 _close_and_reenter를 호출해야 한다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot.exchange = AsyncMock()
+    bot.exchange.get_position = AsyncMock(return_value={
+        "positionAmt": "100",
+        "entryPrice": "0.5",
+        "markPrice": "0.52",
+    })
+    bot._close_and_reenter = AsyncMock()
+    bot.ml_filter = MagicMock()
+    bot.ml_filter.is_model_loaded.return_value = False
+    bot.ml_filter.should_enter.return_value = True
+
+    with patch("src.bot.Indicators") as MockInd:
+        mock_ind = MagicMock()
+        mock_ind.calculate_all.return_value = sample_df
+        mock_ind.get_signal.return_value = "SHORT"  # 현재 LONG 포지션에 반대 시그널
+        MockInd.return_value = mock_ind
+        await bot.process_candle(sample_df)
+
+    bot._close_and_reenter.assert_awaited_once()
+    call_args = bot._close_and_reenter.call_args
+    assert call_args.args[1] == "SHORT"
+
+
+@pytest.mark.asyncio
+async def test_process_candle_passes_raw_signal_to_close_and_reenter_even_if_ml_loaded(config, sample_df):
+    """포지션 보유 시 ML 필터가 로드되어 있어도 process_candle은 raw signal을 _close_and_reenter에 전달한다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot.exchange = AsyncMock()
+    bot.exchange.get_position = AsyncMock(return_value={
+        "positionAmt": "100",
+        "entryPrice": "0.5",
+        "markPrice": "0.52",
+    })
+    bot._close_and_reenter = AsyncMock()
+    bot.ml_filter = MagicMock()
+    bot.ml_filter.is_model_loaded.return_value = True  # 모델 로드됨
+    bot.ml_filter.should_enter.return_value = False    # ML이 차단하더라도
+
+    with patch("src.bot.Indicators") as MockInd:
+        mock_ind = MagicMock()
+        mock_ind.calculate_all.return_value = sample_df
+        mock_ind.get_signal.return_value = "SHORT"
+        MockInd.return_value = mock_ind
+        await bot.process_candle(sample_df)
+
+    # ML 필터가 차단해도 _close_and_reenter는 호출되어야 한다 (ML 재평가는 내부에서)
+    bot._close_and_reenter.assert_awaited_once()
+    call_args = bot._close_and_reenter.call_args
+    assert call_args.args[1] == "SHORT"
+    # process_candle에서 ml_filter.should_enter가 호출되지 않아야 한다
+    bot.ml_filter.should_enter.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_process_candle_fetches_oi_and_funding(config, sample_df):
+    """process_candle()이 OI와 펀딩비를 조회하고 build_features에 전달하는지 확인."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+
+    bot.exchange = AsyncMock()
+    bot.exchange.get_balance = AsyncMock(return_value=1000.0)
+    bot.exchange.get_position = AsyncMock(return_value=None)
+    bot.exchange.place_order = AsyncMock(return_value={"orderId": "1"})
+    bot.exchange.set_leverage = AsyncMock()
+    bot.exchange.get_open_interest = AsyncMock(return_value=5000000.0)
+    bot.exchange.get_funding_rate = AsyncMock(return_value=0.0001)
+
+    # 신호를 LONG으로 강제해 build_features가 반드시 호출되도록 함
+    with patch("src.bot.Indicators") as mock_ind_cls:
+        mock_ind = MagicMock()
+        mock_ind.calculate_all.return_value = sample_df
+        mock_ind.get_signal.return_value = "LONG"
+        mock_ind_cls.return_value = mock_ind
+
+        with patch("src.bot.build_features") as mock_build:
+            from src.ml_features import FEATURE_COLS
+            mock_build.return_value = pd.Series({col: 0.0 for col in FEATURE_COLS})
+            bot.ml_filter.is_model_loaded = MagicMock(return_value=False)
+            # _open_position은 이 테스트의 관심사가 아니므로 mock 처리
+            bot._open_position = AsyncMock()
+            await bot.process_candle(sample_df)
+
+    assert mock_build.called
+    call_kwargs = mock_build.call_args.kwargs
+    assert "oi_change" in call_kwargs
+    assert "funding_rate" in call_kwargs
+
+
+def test_calc_oi_change_first_candle_returns_zero(config):
+    """첫 캔들은 0.0을 반환하고 _prev_oi를 설정한다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+    assert bot._calc_oi_change(5000000.0) == 0.0
+    assert bot._prev_oi == 5000000.0
+
+
+def test_calc_oi_change_api_failure_does_not_corrupt_state(config):
+    """API 실패 시 _fetch_market_microstructure가 _calc_oi_change를 호출하지 않아 상태가 오염되지 않는다."""
+    with patch("src.bot.BinanceFuturesClient"):
+        bot = TradingBot(config)
+    bot._prev_oi = 5000000.0
+    # API 실패 시 _fetch_market_microstructure는 oi_val > 0 체크로 _calc_oi_change를 건너뜀
+    # _calc_oi_change(0.0)을 직접 호출하면 _prev_oi가 0.0으로 오염되는 이전 버그를 재현
+    # 수정 후에는 _fetch_market_microstructure에서 0.0을 직접 반환하므로 이 경로가 없음
+    # 대신 _calc_oi_change가 정상 값에서만 호출되는지 확인
+    result = bot._calc_oi_change(5100000.0)
+    assert abs(result - 0.02) < 1e-6  # (5100000 - 5000000) / 5000000 = 0.02
+    assert bot._prev_oi == 5100000.0

tests/test_data_stream.py

@@ -23,6 +23,7 @@ def test_multi_symbol_stream_get_dataframe_returns_none_when_empty():
 
 def test_multi_symbol_stream_get_dataframe_returns_df_when_full():
     import pandas as pd
+    from src.data_stream import _MIN_CANDLES_FOR_SIGNAL
     stream = MultiSymbolStream(
         symbols=["XRPUSDT", "BTCUSDT", "ETHUSDT"],
         interval="1m",
@@ -32,13 +33,13 @@ def test_multi_symbol_stream_get_dataframe_returns_df_when_full():
         "timestamp": 1000, "open": 1.0, "high": 1.1,
         "low": 0.9, "close": 1.05, "volume": 100.0, "is_closed": True,
     }
-    for i in range(50):
+    for i in range(_MIN_CANDLES_FOR_SIGNAL):
         c = candle.copy()
         c["timestamp"] = 1000 + i
         stream.buffers["xrpusdt"].append(c)
     df = stream.get_dataframe("XRPUSDT")
     assert df is not None
-    assert len(df) == 50
+    assert len(df) == _MIN_CANDLES_FOR_SIGNAL
 
 
 @pytest.mark.asyncio
@@ -62,11 +63,11 @@ async def test_kline_stream_parses_message():
 
 @pytest.mark.asyncio
 async def test_callback_called_on_closed_candle():
-    received = []
+    callback = AsyncMock()
     stream = KlineStream(
         symbol="XRPUSDT",
         interval="1m",
-        on_candle=lambda c: received.append(c),
+        on_candle=callback,
     )
     raw_msg = {
         "k": {
@@ -79,8 +80,8 @@ async def test_callback_called_on_closed_candle():
             "x": True,
         }
     }
-    stream.handle_message(raw_msg)
-    assert len(received) == 1
+    await stream.handle_message(raw_msg)
+    assert callback.call_count == 1
 
 
 @pytest.mark.asyncio

tests/test_dataset_builder.py

@@ -160,3 +160,51 @@ def test_oi_nan_masking_with_zeros():
     feat = _calc_features_vectorized(d, sig)
 
     assert feat["oi_change"].iloc[50:].notna().any(), "실제 OI 값 구간에 유한값이 있어야 함"
+
+
+def test_rs_zero_denominator():
+    """btc_r1=0일 때 RS가 inf/nan이 아닌 0.0이어야 한다 (np.divide 방식 검증)."""
+    import numpy as np
+    import pandas as pd
+    from src.dataset_builder import _calc_features_vectorized, _calc_signals, _calc_indicators
+
+    n = 500
+    np.random.seed(7)
+    # XRP close: 약간의 변동
+    xrp_close = np.cumprod(1 + np.random.randn(n) * 0.001) * 1.0
+    xrp_df = pd.DataFrame({
+        "open":   xrp_close * 0.999,
+        "high":   xrp_close * 1.005,
+        "low":    xrp_close * 0.995,
+        "close":  xrp_close,
+        "volume": np.random.rand(n) * 1000 + 500,
+    })
+    # BTC close: 완전히 고정 → btc_r1 = 0.0
+    btc_close = np.ones(n) * 50000.0
+    btc_df = pd.DataFrame({
+        "open":   btc_close,
+        "high":   btc_close,
+        "low":    btc_close,
+        "close":  btc_close,
+        "volume": np.random.rand(n) * 1000 + 500,
+    })
+    # ETH close: 약간의 변동 (eth_df 없으면 BTC 피처 자체가 계산 안 됨)
+    eth_close = np.cumprod(1 + np.random.randn(n) * 0.001) * 3000.0
+    eth_df = pd.DataFrame({
+        "open":   eth_close * 0.999,
+        "high":   eth_close * 1.005,
+        "low":    eth_close * 0.995,
+        "close":  eth_close,
+        "volume": np.random.rand(n) * 1000 + 500,
+    })
+
+    # _calc_features_vectorized를 직접 호출해 BTC/ETH 피처를 포함한 전체 피처를 검증
+    d = _calc_indicators(xrp_df)
+    signal_arr = _calc_signals(d)
+    feat = _calc_features_vectorized(d, signal_arr, btc_df=btc_df, eth_df=eth_df)
+
+    assert "xrp_btc_rs" in feat.columns, "xrp_btc_rs 컬럼이 있어야 함"
+    assert not feat["xrp_btc_rs"].isin([np.inf, -np.inf]).any(), \
+        "xrp_btc_rs에 inf가 있으면 안 됨"
+    assert not feat["xrp_btc_rs"].isna().all(), \
+        "xrp_btc_rs가 전부 nan이면 안 됨"

tests/test_exchange.py

@@ -25,6 +25,21 @@ def client():
     return c
 
 
+@pytest.fixture
+def exchange():
+    os.environ.update({
+        "BINANCE_API_KEY": "test_key",
+        "BINANCE_API_SECRET": "test_secret",
+        "SYMBOL": "XRPUSDT",
+        "LEVERAGE": "10",
+    })
+    config = Config()
+    c = BinanceFuturesClient.__new__(BinanceFuturesClient)
+    c.config = config
+    c.client = MagicMock()
+    return c
+
+
 @pytest.mark.asyncio
 async def test_set_leverage(config):
     with patch("src.exchange.Client") as MockClient:
@@ -54,3 +69,47 @@ def test_calculate_quantity_zero_balance(client):
     """잔고 0이면 최소 명목금액 기반 수량 반환"""
     qty = client.calculate_quantity(balance=0.0, price=2.5, leverage=10, margin_ratio=0.50)
     assert qty > 0
+
+
+@pytest.mark.asyncio
+async def test_get_open_interest(exchange):
+    """get_open_interest()가 float을 반환하는지 확인."""
+    exchange.client.futures_open_interest = MagicMock(
+        return_value={"openInterest": "123456.789"}
+    )
+    result = await exchange.get_open_interest()
+    assert isinstance(result, float)
+    assert result == pytest.approx(123456.789)
+
+
+@pytest.mark.asyncio
+async def test_get_funding_rate(exchange):
+    """get_funding_rate()가 float을 반환하는지 확인."""
+    exchange.client.futures_mark_price = MagicMock(
+        return_value={"lastFundingRate": "0.0001"}
+    )
+    result = await exchange.get_funding_rate()
+    assert isinstance(result, float)
+    assert result == pytest.approx(0.0001)
+
+
+@pytest.mark.asyncio
+async def test_get_open_interest_error_returns_none(exchange):
+    """API 오류 시 None 반환 확인."""
+    from binance.exceptions import BinanceAPIException
+    exchange.client.futures_open_interest = MagicMock(
+        side_effect=BinanceAPIException(MagicMock(status_code=400), 400, '{"code":-1121,"msg":"Invalid symbol"}')
+    )
+    result = await exchange.get_open_interest()
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_get_funding_rate_error_returns_none(exchange):
+    """API 오류 시 None 반환 확인."""
+    from binance.exceptions import BinanceAPIException
+    exchange.client.futures_mark_price = MagicMock(
+        side_effect=BinanceAPIException(MagicMock(status_code=400), 400, '{"code":-1121,"msg":"Invalid symbol"}')
+    )
+    result = await exchange.get_funding_rate()
+    assert result is None

tests/test_fetch_history.py

@@ -0,0 +1,131 @@
+"""fetch_history.py의 upsert_parquet() 함수 테스트."""
+import pandas as pd
+import numpy as np
+import pytest
+from pathlib import Path
+
+
+def _make_parquet(tmp_path: Path, rows: dict) -> Path:
+    """테스트용 parquet 파일 생성 헬퍼."""
+    df = pd.DataFrame(rows)
+    df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
+    df = df.set_index("timestamp")
+    path = tmp_path / "test.parquet"
+    df.to_parquet(path)
+    return path
+
+
+def test_upsert_fills_zero_oi_with_real_value(tmp_path):
+    """기존 행의 oi_change=0.0이 신규 데이터의 실제 값으로 덮어써진다."""
+    from scripts.fetch_history import upsert_parquet
+
+    existing_path = _make_parquet(tmp_path, {
+        "timestamp": ["2026-01-01 00:00", "2026-01-01 00:15"],
+        "close": [1.0, 1.1],
+        "oi_change": [0.0, 0.0],
+        "funding_rate": [0.0, 0.0],
+    })
+
+    new_df = pd.DataFrame({
+        "close": [1.0, 1.1],
+        "oi_change": [0.05, 0.03],
+        "funding_rate": [0.0001, 0.0001],
+    }, index=pd.to_datetime(["2026-01-01 00:00", "2026-01-01 00:15"], utc=True))
+    new_df.index.name = "timestamp"
+
+    result = upsert_parquet(existing_path, new_df)
+
+    assert result.loc["2026-01-01 00:00+00:00", "oi_change"] == pytest.approx(0.05)
+    assert result.loc["2026-01-01 00:15+00:00", "oi_change"] == pytest.approx(0.03)
+
+
+def test_upsert_appends_new_rows(tmp_path):
+    """신규 타임스탬프 행이 기존 데이터 아래에 추가된다."""
+    from scripts.fetch_history import upsert_parquet
+
+    existing_path = _make_parquet(tmp_path, {
+        "timestamp": ["2026-01-01 00:00"],
+        "close": [1.0],
+        "oi_change": [0.05],
+        "funding_rate": [0.0001],
+    })
+
+    new_df = pd.DataFrame({
+        "close": [1.1],
+        "oi_change": [0.03],
+        "funding_rate": [0.0002],
+    }, index=pd.to_datetime(["2026-01-01 00:15"], utc=True))
+    new_df.index.name = "timestamp"
+
+    result = upsert_parquet(existing_path, new_df)
+
+    assert len(result) == 2
+    assert pd.Timestamp("2026-01-01 00:15", tz="UTC") in result.index
+
+
+def test_upsert_keeps_nonzero_existing_oi(tmp_path):
+    """기존 행의 oi_change가 이미 0이 아니면 덮어쓰지 않는다."""
+    from scripts.fetch_history import upsert_parquet
+
+    existing_path = _make_parquet(tmp_path, {
+        "timestamp": ["2026-01-01 00:00"],
+        "close": [1.0],
+        "oi_change": [0.07],   # 이미 실제 값 존재
+        "funding_rate": [0.0003],
+    })
+
+    new_df = pd.DataFrame({
+        "close": [1.0],
+        "oi_change": [0.05],   # 다른 값으로 덮어쓰려 해도
+        "funding_rate": [0.0001],
+    }, index=pd.to_datetime(["2026-01-01 00:00"], utc=True))
+    new_df.index.name = "timestamp"
+
+    result = upsert_parquet(existing_path, new_df)
+
+    # 기존 값(0.07)이 유지되어야 한다
+    assert result.iloc[0]["oi_change"] == pytest.approx(0.07)
+
+
+def test_upsert_no_existing_file_returns_new_df(tmp_path):
+    """기존 parquet 파일이 없으면 신규 데이터를 그대로 반환한다."""
+    from scripts.fetch_history import upsert_parquet
+
+    nonexistent_path = tmp_path / "nonexistent.parquet"
+    new_df = pd.DataFrame({
+        "close": [1.0, 1.1],
+        "oi_change": [0.05, 0.03],
+        "funding_rate": [0.0001, 0.0001],
+    }, index=pd.to_datetime(["2026-01-01 00:00", "2026-01-01 00:15"], utc=True))
+    new_df.index.name = "timestamp"
+
+    result = upsert_parquet(nonexistent_path, new_df)
+
+    assert len(result) == 2
+    assert result.iloc[0]["oi_change"] == pytest.approx(0.05)
+
+
+def test_upsert_result_is_sorted_by_timestamp(tmp_path):
+    """결과 DataFrame이 timestamp 기준 오름차순 정렬되어 있다."""
+    from scripts.fetch_history import upsert_parquet
+
+    existing_path = _make_parquet(tmp_path, {
+        "timestamp": ["2026-01-01 00:15"],
+        "close": [1.1],
+        "oi_change": [0.0],
+        "funding_rate": [0.0],
+    })
+
+    new_df = pd.DataFrame({
+        "close": [1.0, 1.1, 1.2],
+        "oi_change": [0.05, 0.03, 0.02],
+        "funding_rate": [0.0001, 0.0001, 0.0002],
+    }, index=pd.to_datetime(
+        ["2026-01-01 00:00", "2026-01-01 00:15", "2026-01-01 00:30"], utc=True
+    ))
+    new_df.index.name = "timestamp"
+
+    result = upsert_parquet(existing_path, new_df)
+
+    assert result.index.is_monotonic_increasing
+    assert len(result) == 3

tests/test_ml_features.py

@@ -25,12 +25,12 @@ def test_build_features_with_btc_eth_has_21_features():
     btc_df = _make_df(10, base_price=50000.0)
     eth_df = _make_df(10, base_price=3000.0)
     features = build_features(xrp_df, "LONG", btc_df=btc_df, eth_df=eth_df)
-    assert len(features) == 21
+    assert len(features) == 23
 
 def test_build_features_without_btc_eth_has_13_features():
     xrp_df = _make_df(10, base_price=1.0)
     features = build_features(xrp_df, "LONG")
-    assert len(features) == 13
+    assert len(features) == 15
 
 def test_build_features_btc_ret_1_correct():
     xrp_df = _make_df(10, base_price=1.0)
@@ -49,9 +49,9 @@ def test_build_features_rs_zero_when_btc_ret_zero():
     features = build_features(xrp_df, "LONG", btc_df=btc_df, eth_df=eth_df)
     assert features["xrp_btc_rs"] == 0.0
 
-def test_feature_cols_has_21_items():
+def test_feature_cols_has_23_items():
     from src.ml_features import FEATURE_COLS
-    assert len(FEATURE_COLS) == 21
+    assert len(FEATURE_COLS) == 23
 
 
 def make_df(n=100):
@@ -111,3 +111,30 @@ def test_side_encoding():
     short_feat = build_features(df_ind, signal="SHORT")
     assert long_feat["side"] == 1
     assert short_feat["side"] == 0
+
+
+@pytest.fixture
+def sample_df_with_indicators():
+    from src.indicators import Indicators
+    df = make_df(100)
+    ind = Indicators(df)
+    return ind.calculate_all()
+
+
+def test_build_features_uses_provided_oi_funding(sample_df_with_indicators):
+    """oi_change, funding_rate 파라미터가 제공되면 실제 값이 피처에 반영된다."""
+    feat = build_features(
+        sample_df_with_indicators,
+        signal="LONG",
+        oi_change=0.05,
+        funding_rate=0.0002,
+    )
+    assert feat["oi_change"] == pytest.approx(0.05)
+    assert feat["funding_rate"] == pytest.approx(0.0002)
+
+
+def test_build_features_defaults_to_zero_when_not_provided(sample_df_with_indicators):
+    """oi_change, funding_rate 파라미터 미제공 시 0.0으로 채워진다."""
+    feat = build_features(sample_df_with_indicators, signal="LONG")
+    assert feat["oi_change"] == pytest.approx(0.0)
+    assert feat["funding_rate"] == pytest.approx(0.0)

tests/test_ml_filter.py

@@ -12,13 +12,19 @@ def make_features(side="LONG") -> pd.Series:
 
 
 def test_no_model_file_is_not_loaded(tmp_path):
-    f = MLFilter(model_path=str(tmp_path / "nonexistent.pkl"))
+    f = MLFilter(
+        onnx_path=str(tmp_path / "nonexistent.onnx"),
+        lgbm_path=str(tmp_path / "nonexistent.pkl"),
+    )
     assert not f.is_model_loaded()
 
 
 def test_no_model_should_enter_returns_true(tmp_path):
     """모델 없으면 항상 진입 허용 (폴백)"""
-    f = MLFilter(model_path=str(tmp_path / "nonexistent.pkl"))
+    f = MLFilter(
+        onnx_path=str(tmp_path / "nonexistent.onnx"),
+        lgbm_path=str(tmp_path / "nonexistent.pkl"),
+    )
     features = make_features()
     assert f.should_enter(features) is True
 
@@ -28,7 +34,7 @@ def test_should_enter_above_threshold():
     f = MLFilter(threshold=0.60)
     mock_model = MagicMock()
     mock_model.predict_proba.return_value = np.array([[0.35, 0.65]])
-    f._model = mock_model
+    f._lgbm_model = mock_model
     features = make_features()
     assert f.should_enter(features) is True
 
@@ -38,7 +44,7 @@ def test_should_enter_below_threshold():
     f = MLFilter(threshold=0.60)
     mock_model = MagicMock()
     mock_model.predict_proba.return_value = np.array([[0.55, 0.45]])
-    f._model = mock_model
+    f._lgbm_model = mock_model
     features = make_features()
     assert f.should_enter(features) is False
 
@@ -48,16 +54,18 @@ def test_reload_model(tmp_path):
     import joblib
 
     # 모델 파일이 없는 상태에서 시작
-    model_path = tmp_path / "lgbm_filter.pkl"
-    f = MLFilter(model_path=str(model_path))
+    f = MLFilter(
+        onnx_path=str(tmp_path / "nonexistent.onnx"),
+        lgbm_path=str(tmp_path / "lgbm_filter.pkl"),
+    )
     assert not f.is_model_loaded()
 
-    # _model을 직접 주입해서 is_model_loaded가 True인지 확인
+    # _lgbm_model을 직접 주입해서 is_model_loaded가 True인지 확인
     mock_model = MagicMock()
-    f._model = mock_model
+    f._lgbm_model = mock_model
     assert f.is_model_loaded()
 
-    # reload_model 호출 시 파일이 없으면 _try_load가 _model을 변경하지 않음
-    # (기존 동작 유지 - 파일 없으면 None으로 초기화하지 않음)
+    # reload_model은 항상 _lgbm_model/_onnx_session을 초기화 후 재로드한다.
+    # 파일이 없으면 None으로 리셋되어 폴백 상태가 된다.
     f.reload_model()
-    assert f.is_model_loaded()  # mock_model이 유지됨
+    assert not f.is_model_loaded()  # 파일 없으므로 폴백 상태

tests/test_mlx_filter.py

@@ -10,23 +10,11 @@ mlx = pytest.importorskip("mlx.core", reason="MLX 미설치")
 
 
 def _make_X(n: int = 4) -> pd.DataFrame:
+    from src.ml_features import FEATURE_COLS
     rng = np.random.default_rng(0)
     return pd.DataFrame(
-        {
-            "rsi": rng.uniform(20, 80, n),
-            "macd_hist": rng.uniform(-0.1, 0.1, n),
-            "bb_pct": rng.uniform(0, 1, n),
-            "ema_align": rng.choice([-1.0, 0.0, 1.0], n),
-            "stoch_k": rng.uniform(0, 100, n),
-            "stoch_d": rng.uniform(0, 100, n),
-            "atr_pct": rng.uniform(0.001, 0.05, n),
-            "vol_ratio": rng.uniform(0.5, 3.0, n),
-            "ret_1": rng.uniform(-0.01, 0.01, n),
-            "ret_3": rng.uniform(-0.02, 0.02, n),
-            "ret_5": rng.uniform(-0.03, 0.03, n),
-            "signal_strength": rng.integers(0, 6, n).astype(float),
-            "side": rng.choice([0.0, 1.0], n),
-        }
+        rng.uniform(-1.0, 1.0, (n, len(FEATURE_COLS))).astype(np.float32),
+        columns=FEATURE_COLS,
     )
 
 
@@ -41,9 +29,10 @@ def test_mlx_gpu_device():
 def test_mlx_filter_predict_shape_untrained():
     """학습 전에도 predict_proba가 (N,) 형태를 반환해야 한다."""
     from src.mlx_filter import MLXFilter
+    from src.ml_features import FEATURE_COLS
 
     X = _make_X(4)
-    model = MLXFilter(input_dim=13, hidden_dim=32)
+    model = MLXFilter(input_dim=len(FEATURE_COLS), hidden_dim=32)
     proba = model.predict_proba(X)
     assert proba.shape == (4,)
     assert np.all((proba >= 0.0) & (proba <= 1.0))
@@ -52,12 +41,13 @@ def test_mlx_filter_predict_shape_untrained():
 def test_mlx_filter_fit_and_predict():
     """학습 후 predict_proba가 유효한 확률값을 반환해야 한다."""
     from src.mlx_filter import MLXFilter
+    from src.ml_features import FEATURE_COLS
 
     n = 100
     X = _make_X(n)
     y = pd.Series(np.random.randint(0, 2, n))
 
-    model = MLXFilter(input_dim=13, hidden_dim=32, epochs=5, batch_size=32)
+    model = MLXFilter(input_dim=len(FEATURE_COLS), hidden_dim=32, epochs=5, batch_size=32)
     model.fit(X, y)
     proba = model.predict_proba(X)
 
@@ -93,12 +83,13 @@ def test_fit_with_nan_features():
 def test_mlx_filter_save_load(tmp_path):
     """저장 후 로드한 모델이 동일한 예측값을 반환해야 한다."""
     from src.mlx_filter import MLXFilter
+    from src.ml_features import FEATURE_COLS
 
     n = 50
     X = _make_X(n)
     y = pd.Series(np.random.randint(0, 2, n))
 
-    model = MLXFilter(input_dim=13, hidden_dim=32, epochs=3, batch_size=32)
+    model = MLXFilter(input_dim=len(FEATURE_COLS), hidden_dim=32, epochs=3, batch_size=32)
     model.fit(X, y)
     proba_before = model.predict_proba(X)