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
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
import requests
|
|
import json
|
|
import db
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
DB_PATH = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
|
|
URL = "https://api.hyperliquid.xyz/info"
|
|
|
|
def fetch_historical_candles(coin, start_ms, end_ms, interval="1m"):
|
|
"""Fetch historical candles using the raw HTTP API."""
|
|
candles = []
|
|
current_start = start_ms
|
|
while current_start < end_ms:
|
|
payload = {
|
|
"type": "candleSnapshot",
|
|
"req": {
|
|
"coin": coin,
|
|
"interval": interval,
|
|
"startTime": current_start,
|
|
"endTime": end_ms
|
|
}
|
|
}
|
|
resp = requests.post(URL, json=payload)
|
|
batch = resp.json()
|
|
if not batch:
|
|
break
|
|
for candle in batch:
|
|
candle['coin'] = coin
|
|
candles.append(candle)
|
|
last_ts = batch[-1]['t']
|
|
if last_ts < current_start:
|
|
break
|
|
current_start = last_ts + 1
|
|
time.sleep(0.5)
|
|
return candles
|
|
|
|
def write_candles_to_db(coin, candles, interval="1m"):
|
|
"""Write candles to the database."""
|
|
table_name = db.sanitize_table_name(coin, interval)
|
|
conn = db.get_connection()
|
|
db.create_candle_table(conn, table_name)
|
|
records = []
|
|
for candle in candles:
|
|
record = (
|
|
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')
|
|
)
|
|
records.append(record)
|
|
db.upsert_candles(conn, table_name, records)
|
|
conn.close()
|
|
|
|
def get_last_timestamp(coin):
|
|
"""Get the most recent timestamp from the database."""
|
|
table_name = db.sanitize_table_name(coin, "1m")
|
|
conn = db.get_connection()
|
|
try:
|
|
return db.get_last_timestamp(conn, table_name)
|
|
except:
|
|
return None
|
|
finally:
|
|
conn.close()
|
|
|
|
coins = ["mkts:USTECH", "xyz:XYZ100"]
|
|
now_ms = int(time.time() * 1000)
|
|
seven_days_ms = 7 * 24 * 60 * 60 * 1000
|
|
|
|
for coin in coins:
|
|
for tf in ["1m", "1d"]:
|
|
start_ts = now_ms - seven_days_ms
|
|
if start_ts >= now_ms:
|
|
print(f"{coin} ({tf}): Already up to date")
|
|
continue
|
|
|
|
print(f"{coin} ({tf}): Fetching historical candles from {datetime.fromtimestamp(start_ts/1000, tz=timezone.utc)} to {datetime.fromtimestamp(now_ms/1000, tz=timezone.utc)}...")
|
|
candles = fetch_historical_candles(coin, start_ts, now_ms, interval=tf)
|
|
print(f"{coin} ({tf}): Fetched {len(candles)} candles")
|
|
write_candles_to_db(coin, candles, interval=tf)
|
|
print(f"{coin} ({tf}): Written to database")
|
|
|
|
print("Done!")
|