Batch bulk DB writes (execute_batch): unfreeze the discovery step
The live loop froze ~40s every discovery pass. py-spy showed the main thread stuck in upsert_daily_bars -> PgConnection.executemany, which used psycopg2's cur.executemany — ONE INSERT round-trip per row. A discovery pass upserts thousands of daily bars across the universe, so over the LAN Postgres that's thousands of serial round-trips. While frozen the loop emits nothing and can't service the trigger/exit cadence. Switch executemany to psycopg2.extras.execute_batch (page_size=200): many statements per round-trip, same SQL, same ON CONFLICT upsert semantics. Benchmarked on the live DB: 771-row upsert 1.87s -> 0.06s (~31x); a real discovery pass now finishes in one cycle (ingest 6803 rows, no stall). Helps every bulk writer (daily/minute bars, scanner snapshots). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
1 parent
1acf77adb1
commit
5ed038feb4
1 file changed
+6
-1
+6
-1
@@ -37,6 +37,7 @@ from typing import Any
|
||||
|
||||
import pandas as pd
|
||||
import psycopg2
|
||||
from psycopg2.extras import execute_batch
|
||||
|
||||
SCHEMA_SQL_PATH = Path(__file__).with_name("pg_schema.sql")
|
||||
|
||||
@@ -138,7 +139,11 @@ class PgConnection:
|
||||
|
||||
def go():
|
||||
cur = self._conn.cursor()
|
||||
cur.executemany(_translate(sql), seq)
|
||||
# execute_batch sends page_size statements per round-trip; plain
|
||||
# psycopg2 executemany sends ONE per row. Over the remote LAN
|
||||
# Postgres that was ~771 round-trips for a daily-bar upsert and
|
||||
# froze the live loop's discovery step for ~40s each pass.
|
||||
execute_batch(cur, _translate(sql), seq, page_size=200)
|
||||
cur.close()
|
||||
return self
|
||||
return self._run(go)
|
||||
|
||||
Reference in new issue
Block a user