fix: resolve critical/important bugs from code review (#1,#2,#4,#5,#6,#8)

- #1: OI division by zero — already fixed (prev_oi == 0.0 guard exists)
- #2: cumulative trade count used max() instead of sum(), breaking ML trigger
- #4: fetch_history API calls now retry 3x with exponential backoff
- #5: parquet upsert now deduplicates timestamps before sort
- #6: record_pnl() is now async with Lock for multi-symbol safety
- #8: exit_price=0.0 skips close handling with warning log

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

docs/plans/2026-03-07-code-review-improvements.md

@@ -1,7 +1,7 @@
 # 코드 리뷰 개선 사항
 
 **날짜**: 2026-03-07
-**상태**: 대기
+**상태**: 부분 완료 (#1 기수정, #2/#4/#5/#6/#8 완료, #9 보류, #3/#7/#10~13 다음 스프린트)
 
 ## 목표
 

scripts/fetch_history.py

@@ -44,12 +44,23 @@ async def _fetch_klines_with_client(
     start_ts = int((datetime.now(timezone.utc) - timedelta(days=days)).timestamp() * 1000)
     all_klines = []
     while True:
-        klines = await client.futures_klines(
-            symbol=symbol,
-            interval=interval,
-            startTime=start_ts,
-            limit=1500,
-        )
+        for attempt in range(3):
+            try:
+                klines = await client.futures_klines(
+                    symbol=symbol,
+                    interval=interval,
+                    startTime=start_ts,
+                    limit=1500,
+                )
+                break
+            except Exception as e:
+                if attempt < 2:
+                    wait = 2 ** (attempt + 1)
+                    print(f"  [{symbol}] API 오류 ({e}), {wait}초 후 재시도 ({attempt+1}/3)")
+                    await asyncio.sleep(wait)
+                else:
+                    print(f"  [{symbol}] API 3회 실패, 수집 중단: {e}")
+                    raise
         if not klines:
             break
         all_klines.extend(klines)
@@ -311,6 +322,7 @@ def upsert_parquet(path: "Path | str", new_df: pd.DataFrame) -> pd.DataFrame:
         if col in existing.columns:
             existing[col] = existing[col].fillna(0.0)
 
+    existing = existing[~existing.index.duplicated(keep='last')]
     return existing.sort_index()
 
 

scripts/weekly_report.py

@@ -418,7 +418,7 @@ def generate_report(
         for rpath in sorted(rdir.glob("report_*.json")):
             try:
                 prev = json.loads(rpath.read_text())
-                cumulative = max(cumulative, prev.get("live_trades", {}).get("count", 0))
+                cumulative += prev.get("live_trades", {}).get("count", 0)
             except (json.JSONDecodeError, KeyError):
                 pass
 

src/risk_manager.py

@@ -52,9 +52,10 @@ class RiskManager:
             self.daily_pnl += pnl
             logger.info(f"포지션 종료: {symbol}, PnL={pnl:+.4f}, 누적={self.daily_pnl:+.4f}")
 
-    def record_pnl(self, pnl: float):
-        self.daily_pnl += pnl
-        logger.info(f"오늘 누적 PnL: {self.daily_pnl:.4f} USDT")
+    async def record_pnl(self, pnl: float):
+        async with self._lock:
+            self.daily_pnl += pnl
+            logger.info(f"오늘 누적 PnL: {self.daily_pnl:.4f} USDT")
 
     def reset_daily(self):
         """매일 자정 초기화"""

src/user_data_stream.py

@@ -94,6 +94,13 @@ class UserDataStream:
         net_pnl    = realized_pnl - commission
         exit_price = float(order.get("ap", "0"))
 
+        if exit_price == 0.0:
+            logger.warning(
+                f"[{self._symbol}] 청산 이벤트에서 exit_price=0.0 — "
+                f"ap 필드 누락 가능. 청산 처리 스킵 (rp={realized_pnl:+.4f})"
+            )
+            return
+
         if order_type == "TAKE_PROFIT_MARKET":
             close_reason = "TP"
         elif order_type == "STOP_MARKET":