7d702e9cbd
- Add db.py PostgreSQL abstraction layer (connection, upsert, table mgmt) - Replace sqlite3 with psycopg2 in: live_candle_fetcher, resampler, data_fetcher, fetch_history, import_csv, indicators, base_strategy - Sanitize table names (colons -> underscores) for PostgreSQL compat - Replace INSERT OR REPLACE with ON CONFLICT upserts - Replace pandas to_sql() with batch upsert_candles() - Add scripts: resampler_loop, gap_detector, backup_runner, cron_scheduler - Add migrate_sqlite_to_pg.py for one-time data migration - Add Dockerfile, docker-compose.yml, supervisord.conf - Add postgres/postgresql.conf tuned for 4GB RAM (Synology DS1513+) - Add .dockerignore, .env.docker.example, secrets template - Update requirements.txt (psycopg2-binary), .gitignore - Add MIGRATION_PLAN.md with full plan and todo list
138 lines
4.1 KiB
Python
138 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Gap Detector
|
|
|
|
Detects missing 1-minute candle data in the PostgreSQL database and
|
|
backfills gaps by fetching historical data from the Hyperliquid HTTP API.
|
|
|
|
Designed to run as a periodic cron job (hourly) inside the Docker container.
|
|
"""
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pandas as pd
|
|
from hyperliquid.info import Info
|
|
from hyperliquid.utils import constants
|
|
|
|
from logging_utils import setup_logging
|
|
from db import get_connection, sanitize_table_name, upsert_candles
|
|
|
|
WATCHED_COINS = [
|
|
"BTC", "ETH", "SOL", "BNB", "HYPE", "SUI",
|
|
"xyz:BRENTOIL", "xyz:CL", "xyz:GOLD", "xyz:SILVER",
|
|
"mkts:USTECH", "xyz:XYZ100"
|
|
]
|
|
|
|
|
|
def detect_and_fill_gaps(coin, conn):
|
|
"""Detect gaps in the 1m data for a coin and backfill them."""
|
|
table_name = sanitize_table_name(coin, "1m")
|
|
|
|
now = datetime.now(timezone.utc)
|
|
start = now - timedelta(hours=24)
|
|
|
|
query = f'SELECT timestamp_ms FROM "{table_name}" WHERE timestamp_ms >= %s ORDER BY timestamp_ms'
|
|
df = pd.read_sql(query, conn, params=(int(start.timestamp() * 1000),))
|
|
|
|
if df.empty:
|
|
logging.info(f"No data for {coin} in the last 24 hours, skipping gap detection")
|
|
return
|
|
|
|
existing_timestamps = set(df['timestamp_ms'].tolist())
|
|
|
|
# Generate expected timestamps (every minute)
|
|
expected_timestamps = set()
|
|
current = start
|
|
while current <= now:
|
|
expected_timestamps.add(int(current.timestamp() * 1000))
|
|
current += timedelta(minutes=1)
|
|
|
|
gaps = expected_timestamps - existing_timestamps
|
|
|
|
if not gaps:
|
|
logging.info(f"No gaps found for {coin}")
|
|
return
|
|
|
|
logging.info(f"Found {len(gaps)} gaps for {coin}, backfilling...")
|
|
|
|
# Find contiguous gap ranges
|
|
sorted_gaps = sorted(gaps)
|
|
gap_ranges = []
|
|
gap_start = sorted_gaps[0]
|
|
gap_end = sorted_gaps[0]
|
|
|
|
for ts in sorted_gaps[1:]:
|
|
if ts == gap_end + 60000:
|
|
gap_end = ts
|
|
else:
|
|
gap_ranges.append((gap_start, gap_end + 60000))
|
|
gap_start = ts
|
|
gap_end = ts
|
|
gap_ranges.append((gap_start, gap_end + 60000))
|
|
|
|
info = Info(constants.MAINNET_API_URL, skip_ws=True)
|
|
|
|
for gap_start_ms, gap_end_ms in gap_ranges:
|
|
logging.info(
|
|
f"Backfilling gap for {coin}: "
|
|
f"{datetime.fromtimestamp(gap_start_ms/1000, tz=timezone.utc)} "
|
|
f"to {datetime.fromtimestamp(gap_end_ms/1000, tz=timezone.utc)}"
|
|
)
|
|
|
|
current_start = gap_start_ms
|
|
while current_start < gap_end_ms:
|
|
try:
|
|
batch = info.candles_snapshot(coin, "1m", current_start, gap_end_ms)
|
|
if not batch:
|
|
break
|
|
|
|
records = []
|
|
for candle in batch:
|
|
records.append((
|
|
datetime.fromtimestamp(candle['t'] / 1000, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S'),
|
|
candle['t'],
|
|
candle.get('o'), candle.get('h'), candle.get('l'), candle.get('c'),
|
|
candle.get('v'), candle.get('n')
|
|
))
|
|
|
|
upsert_candles(conn, table_name, records)
|
|
|
|
last_ts = batch[-1]['t']
|
|
if last_ts < current_start:
|
|
break
|
|
current_start = last_ts + 1
|
|
time.sleep(0.5)
|
|
except Exception as e:
|
|
logging.error(f"Error backfilling gap for {coin}: {e}")
|
|
break
|
|
|
|
logging.info(f"Gap backfilling complete for {coin}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Detect and fill gaps in 1m candle data.")
|
|
parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
|
|
args = parser.parse_args()
|
|
|
|
setup_logging(args.log_level, 'GapDetector')
|
|
|
|
conn = get_connection()
|
|
|
|
for coin in WATCHED_COINS:
|
|
try:
|
|
detect_and_fill_gaps(coin, conn)
|
|
except Exception as e:
|
|
logging.error(f"Error detecting gaps for {coin}: {e}")
|
|
|
|
conn.close()
|
|
logging.info("Gap detection complete!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|