知行多空线指标接入选股与事件回测；大盘总览走 quicksync 统一入口；candles 批量 upsert 分批防 asyncpg 参数超限

backend/app/screener/market_sync.py

@@ -245,18 +245,21 @@ async def _upsert_candle_day(session: AsyncSession, rows: list[dict], listed: se
     ]
     if not batch:
         return
-    stmt = pg_insert(Candle).values(batch)
-    stmt = stmt.on_conflict_do_update(
-        index_elements=["symbol", "timeframe", "ts"],
-        set_={
-            "open": stmt.excluded.open, "high": stmt.excluded.high,
-            "low": stmt.excluded.low, "close": stmt.excluded.close,
-            "volume": stmt.excluded.volume,
-            "amount": func.coalesce(Candle.amount, stmt.excluded.amount),
-        },
-    )
-    await session.execute(stmt)
-    await session.commit()
+    # on_conflict 语句整批渲染为占位符（非 executemany），asyncpg 单语句参数上限 32766，
+    # 10 列 x 3000 行 = 30000 参数留出余量
+    for i in range(0, len(batch), 3000):
+        stmt = pg_insert(Candle).values(batch[i : i + 3000])
+        stmt = stmt.on_conflict_do_update(
+            index_elements=["symbol", "timeframe", "ts"],
+            set_={
+                "open": stmt.excluded.open, "high": stmt.excluded.high,
+                "low": stmt.excluded.low, "close": stmt.excluded.close,
+                "volume": stmt.excluded.volume,
+                "amount": func.coalesce(Candle.amount, stmt.excluded.amount),
+            },
+        )
+        await session.execute(stmt)
+        await session.commit()
 
 
 async def _run_sync(days: int, force: bool) -> None: