Release v1.3.1 reliability improvements

scanner.py

@@ -12,6 +12,7 @@ from typing import Any
 from zoneinfo import ZoneInfo
 
 import aiohttp
+from filelock import FileLock, Timeout as FileLockTimeout
 from dotenv import load_dotenv
 
 
@@ -39,13 +40,14 @@ CRYPTO_BODY_RATIO_FILTER_ENABLED = (
 )
 CRYPTO_MIN_BODY_RATIO = Decimal(os.getenv("CRYPTO_MIN_BODY_RATIO", str(MIN_BODY_RATIO)))
 TRADFI_BODY_RATIO_FILTER_ENABLED = (
-    os.getenv("TRADFI_BODY_RATIO_FILTER_ENABLED", str(BODY_RATIO_FILTER_ENABLED)).lower() == "true"
+    os.getenv("TRADFI_BODY_RATIO_FILTER_ENABLED", "true").lower() == "true"
 )
 TRADFI_MIN_BODY_RATIO = Decimal(os.getenv("TRADFI_MIN_BODY_RATIO", str(MIN_BODY_RATIO)))
 KLINES_RETENTION_PER_SYMBOL = int(os.getenv("KLINES_RETENTION_PER_SYMBOL", "500"))
 SIGNAL_RETENTION_DAYS = int(os.getenv("SIGNAL_RETENTION_DAYS", "90"))
 DISCORD_ENABLED = os.getenv("DISCORD_ENABLED", "false").lower() == "true"
 DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL", "").strip()
+SCAN_LOCK_PATH = Path(os.getenv("SCAN_LOCK_PATH", "data/scanner.lock"))
 
 MARKET_CRYPTO = "CRYPTO"
 MARKET_TRADFI = "TRADFI"
@@ -71,6 +73,7 @@ class ScanResult:
     symbols_failed: int
     updated_klines: int
     signals_created: int
+    current_signals: int
     elapsed_seconds: float
     error_summary: str | None = None
 
@@ -652,6 +655,21 @@ def insert_scan_run(
     conn.commit()
 
 
+def count_current_signals(conn: sqlite3.Connection, market_type: str) -> int:
+    row = conn.execute(
+        """
+        SELECT COUNT(*) AS count
+        FROM signals
+        WHERE market_type = ?
+          AND open_time = (
+              SELECT MAX(open_time) FROM klines WHERE market_type = ?
+          )
+        """,
+        (market_type, market_type),
+    ).fetchone()
+    return int(row["count"]) if row else 0
+
+
 async def fetch_missing_closed_klines(
     session: aiohttp.ClientSession,
     symbol: str,
@@ -860,6 +878,7 @@ async def run_scan(market_type: str = MARKET_CRYPTO) -> ScanResult:
         elapsed_seconds=elapsed,
         latest_target_open_time=expected_open_time,
     )
+    current_signals = count_current_signals(conn, market_type)
     conn.close()
     speed = len(symbols) / elapsed if elapsed > 0 else 0
     logger.info(
@@ -891,6 +910,7 @@ async def run_scan(market_type: str = MARKET_CRYPTO) -> ScanResult:
         symbols_failed=failures,
         updated_klines=updated,
         signals_created=signals,
+        current_signals=current_signals,
         elapsed_seconds=elapsed,
     )
 
@@ -909,12 +929,12 @@ async def send_discord_summary(results: list[ScanResult], elapsed_seconds: float
         market_name = "Crypto" if result.market_type == MARKET_CRYPTO else "TradFi"
         status = status_labels.get(result.status, result.status)
         lines.append(
-            f"{market_name}：{result.signals_created} 个信号 · {status} · "
+            f"{market_name}：{result.current_signals} 个信号 · {status} · "
             f"失败 {result.symbols_failed}"
         )
     lines.extend(
         [
-            f"合计：{sum(result.signals_created for result in results)} 个信号",
+            f"合计：{sum(result.current_signals for result in results)} 个信号",
             f"耗时：{elapsed_seconds:.1f} 秒",
             f"时间：{madrid_time.strftime('%Y-%m-%d %H:%M')} 马德里",
         ]
@@ -955,6 +975,7 @@ async def run_all_markets() -> list[ScanResult]:
                     symbols_failed=0,
                     updated_klines=0,
                     signals_created=0,
+                    current_signals=0,
                     elapsed_seconds=0,
                     error_summary=str(exc),
                 )
@@ -963,6 +984,20 @@ async def run_all_markets() -> list[ScanResult]:
     return results
 
 
+def run_cli(selected: str) -> None:
+    SCAN_LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
+    lock = FileLock(str(SCAN_LOCK_PATH))
+    try:
+        with lock.acquire(timeout=0):
+            if selected == "all":
+                asyncio.run(run_all_markets())
+            else:
+                market_type = MARKET_TRADFI if selected == "tradfi" else MARKET_CRYPTO
+                asyncio.run(run_scan(market_type))
+    except FileLockTimeout:
+        logger.warning("Scan skipped: another scanner process is already running")
+
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Scan Binance ATR signals")
     parser.add_argument(
@@ -972,8 +1007,4 @@ if __name__ == "__main__":
         help="Market to scan: crypto, tradfi, or all",
     )
     args = parser.parse_args()
-    if args.market == "all":
-        asyncio.run(run_all_markets())
-    else:
-        selected_market = MARKET_TRADFI if args.market == "tradfi" else MARKET_CRYPTO
-        asyncio.run(run_scan(selected_market))
+    run_cli(args.market)