Add scanner diagnostics for zero-candidate trade stalls

docs/issues/ISSUE-2026-02-17-no-trades-zero-candidates.md

@@ -0,0 +1,29 @@
+# Issue: Realtime 모드에서 거래가 지속적으로 0건
+
+## Summary
+`realtime` 실행 중 주문 단계까지 진입하지 못하고, 스캐너 단계에서 후보가 0건으로 반복 종료된다.
+
+## Observed
+- 로그에서 반복적으로 `Smart Scanner: No candidates ... — no trades` 출력
+- 해외 시장에서 `Overseas ranking endpoint unavailable (404)` 다수 발생
+- fallback 심볼 스캔도 `0 candidates`로 종료
+- `data/trade_logs.db` 기준 최근 구간에 `BUY/SELL` 없음
+
+## Impact
+- 매매 전략 품질과 무관하게 주문 경로가 실행되지 않아 실질 거래 불가
+- 장애 원인을 로그만으로 즉시 분해하기 어려움
+
+## Root-Cause Hypothesis
+- 스캐너 필터(가격/변동성) 단계에서 대부분 탈락
+- 해외 랭킹 API 불가 시 입력 유니버스가 빈 상태가 되어 후보 생성 실패
+- 기존 로그는 최종 결과(0 candidates)만 보여 원인별 분해가 어려움
+
+## Acceptance Criteria
+- 스캔 1회마다 탈락 사유가 구조화되어 로그에 남아야 함
+- 국내/해외(랭킹/폴백) 경로 모두 동일한 진단 지표를 제공해야 함
+- 운영자가 로그만 보고 `왜 0 candidates인지`를 즉시 판단 가능해야 함
+
+## Scope
+- 이번 이슈는 **진단 가능성 개선(Observability)** 에 한정
+- 후보 생성 전략 변경(기본 유니버스 강제 추가 등)은 별도 이슈로 분리
+

docs/prs/PR-2026-02-17-scanner-diagnostics-for-zero-candidates.md

@@ -0,0 +1,32 @@
+# PR: Smart Scanner 진단 로그 추가 (0 candidates 원인 분해)
+
+## Linked Issue
+- `docs/issues/ISSUE-2026-02-17-no-trades-zero-candidates.md`
+
+## What Changed
+- `src/analysis/smart_scanner.py`에 스캔 진단 카운터 추가
+- 국내 스캔 진단 로그 추가
+- 해외 랭킹 스캔 진단 로그 추가
+- 해외 fallback 심볼 스캔 진단 로그 추가
+
+## Diagnostics Keys
+- `total_rows`
+- `missing_code`
+- `invalid_price`
+- `low_volatility`
+- `connection_error` (해당 경로에서만)
+- `unexpected_error` (해당 경로에서만)
+- `qualified`
+
+## Expected Log Examples
+- `Domestic scan diagnostics: {...}`
+- `Overseas ranking scan diagnostics for US_NASDAQ: {...}`
+- `Overseas fallback scan diagnostics for US_NYSE: {...}`
+
+## Out of Scope
+- 해외 랭킹 404 시 기본 심볼 유니버스 강제 주입
+- 국내 경로 fallback 정책 변경
+
+## Validation
+- `.venv/bin/python -m py_compile src/analysis/smart_scanner.py`
+

src/analysis/smart_scanner.py

@@ -128,6 +128,16 @@ class SmartVolatilityScanner:
         if not fluct_rows:
             return []
 
+        diagnostics: dict[str, int | float] = {
+            "total_rows": len(fluct_rows),
+            "missing_code": 0,
+            "invalid_price": 0,
+            "low_volatility": 0,
+            "connection_error": 0,
+            "unexpected_error": 0,
+            "qualified": 0,
+        }
+
         volume_rank_bonus: dict[str, float] = {}
         for idx, row in enumerate(volume_rows):
             code = _extract_stock_code(row)
@@ -139,6 +149,7 @@ class SmartVolatilityScanner:
         for stock in fluct_rows:
             stock_code = _extract_stock_code(stock)
             if not stock_code:
+                diagnostics["missing_code"] += 1
                 continue
 
             try:
@@ -168,7 +179,11 @@ class SmartVolatilityScanner:
                             volume_ratio = max(volume_ratio, volume / prev_day_volume)
 
                 volatility_pct = max(abs(change_rate), intraday_range_pct)
-                if price <= 0 or volatility_pct < 0.8:
+                if price <= 0:
+                    diagnostics["invalid_price"] += 1
+                    continue
+                if volatility_pct < 0.8:
+                    diagnostics["low_volatility"] += 1
                     continue
 
                 volatility_score = min(volatility_pct / 10.0, 1.0) * 85.0
@@ -189,14 +204,22 @@ class SmartVolatilityScanner:
                         score=score,
                     )
                 )
+                diagnostics["qualified"] += 1
 
             except ConnectionError as exc:
+                diagnostics["connection_error"] += 1
                 logger.warning("Failed to analyze %s: %s", stock_code, exc)
                 continue
             except Exception as exc:
+                diagnostics["unexpected_error"] += 1
                 logger.error("Unexpected error analyzing %s: %s", stock_code, exc)
                 continue
 
+        logger.info(
+            "Domestic scan diagnostics: %s (volatility_threshold=0.8, top_n=%d)",
+            diagnostics,
+            self.top_n,
+        )
         logger.info("Domestic ranking scan found %d candidates", len(candidates))
         candidates.sort(key=lambda c: c.score, reverse=True)
         return candidates[: self.top_n]
@@ -242,6 +265,14 @@ class SmartVolatilityScanner:
         if not fluct_rows:
             return []
 
+        diagnostics: dict[str, int | float] = {
+            "total_rows": len(fluct_rows),
+            "missing_code": 0,
+            "invalid_price": 0,
+            "low_volatility": 0,
+            "qualified": 0,
+        }
+
         volume_rank_bonus: dict[str, float] = {}
         try:
             volume_rows = await self.overseas_broker.fetch_overseas_rankings(
@@ -266,6 +297,7 @@ class SmartVolatilityScanner:
         for row in fluct_rows:
             stock_code = _extract_stock_code(row)
             if not stock_code:
+                diagnostics["missing_code"] += 1
                 continue
 
             price = _extract_last_price(row)
@@ -275,7 +307,11 @@ class SmartVolatilityScanner:
             volatility_pct = max(abs(change_rate), intraday_range_pct)
 
             # Volatility-first filter (not simple gainers/value ranking).
-            if price <= 0 or volatility_pct < 0.8:
+            if price <= 0:
+                diagnostics["invalid_price"] += 1
+                continue
+            if volatility_pct < 0.8:
+                diagnostics["low_volatility"] += 1
                 continue
 
             volatility_score = min(volatility_pct / 10.0, 1.0) * 85.0
@@ -295,7 +331,14 @@ class SmartVolatilityScanner:
                     score=score,
                 )
             )
+            diagnostics["qualified"] += 1
 
+        logger.info(
+            "Overseas ranking scan diagnostics for %s: %s (volatility_threshold=0.8, top_n=%d)",
+            market.code,
+            diagnostics,
+            self.top_n,
+        )
         if candidates:
             logger.info(
                 "Overseas ranking scan found %d candidates for %s",
@@ -320,6 +363,14 @@ class SmartVolatilityScanner:
             len(symbols),
             market.name,
         )
+        diagnostics: dict[str, int | float] = {
+            "total_rows": len(symbols),
+            "invalid_price": 0,
+            "low_volatility": 0,
+            "connection_error": 0,
+            "unexpected_error": 0,
+            "qualified": 0,
+        }
         candidates: list[ScanCandidate] = []
         for stock_code in symbols:
             try:
@@ -333,7 +384,11 @@ class SmartVolatilityScanner:
                 intraday_range_pct = _extract_intraday_range_pct(output, price)
                 volatility_pct = max(abs(change_rate), intraday_range_pct)
 
-                if price <= 0 or volatility_pct < 0.8:
+                if price <= 0:
+                    diagnostics["invalid_price"] += 1
+                    continue
+                if volatility_pct < 0.8:
+                    diagnostics["low_volatility"] += 1
                     continue
 
                 score = min(volatility_pct / 10.0, 1.0) * 100.0
@@ -351,10 +406,19 @@ class SmartVolatilityScanner:
                         score=score,
                     )
                 )
+                diagnostics["qualified"] += 1
             except ConnectionError as exc:
+                diagnostics["connection_error"] += 1
                 logger.warning("Failed to analyze overseas %s: %s", stock_code, exc)
             except Exception as exc:
+                diagnostics["unexpected_error"] += 1
                 logger.error("Unexpected error analyzing overseas %s: %s", stock_code, exc)
+        logger.info(
+            "Overseas fallback scan diagnostics for %s: %s (volatility_threshold=0.8, top_n=%d)",
+            market.code,
+            diagnostics,
+            self.top_n,
+        )
         logger.info(
             "Overseas symbol fallback scan found %d candidates for %s",
             len(candidates),