Migrate data pipeline from SQLite to PostgreSQL + Docker setup
- 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
This commit is contained in:
DiTus
88
resampler.py
88
resampler.py
@ -2,7 +2,7 @@ import argparse
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import sqlite3
|
||||
import db
|
||||
import pandas as pd
|
||||
import json
|
||||
from datetime import datetime, timezone, timedelta
|
||||
@ -19,7 +19,7 @@ class Resampler:
|
||||
|
||||
def __init__(self, log_level: str, coins: list, timeframes: dict):
|
||||
setup_logging(log_level, 'Resampler')
|
||||
self.db_path = os.path.join("_data", "market_data.db")
|
||||
self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
|
||||
self.status_file_path = os.path.join("_data", "resampling_status.json")
|
||||
self.coins_to_process = coins
|
||||
self.timeframes = timeframes
|
||||
@ -37,58 +37,16 @@ class Resampler:
|
||||
|
||||
def _ensure_tables_exist(self):
|
||||
"""
|
||||
Ensures all resampled tables exist with a PRIMARY KEY on timestamp_ms.
|
||||
Attempts to migrate existing tables if the schema is incorrect.
|
||||
Ensures all resampled tables exist with the correct schema.
|
||||
Uses db.create_candle_table() which is idempotent.
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
for coin in self.coins_to_process:
|
||||
for tf_name in self.timeframes.keys():
|
||||
table_name = f"{coin}_{tf_name}"
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(f"PRAGMA table_info('{table_name}')")
|
||||
columns = cursor.fetchall()
|
||||
if columns:
|
||||
# --- FIX: Check for the correct PRIMARY KEY on timestamp_ms ---
|
||||
pk_found = any(col[1] == 'timestamp_ms' and col[5] == 1 for col in columns)
|
||||
if not pk_found:
|
||||
logging.warning(f"Schema migration needed for table '{table_name}'.")
|
||||
try:
|
||||
conn.execute(f'ALTER TABLE "{table_name}" RENAME TO "{table_name}_old"')
|
||||
self._create_resampled_table(conn, table_name)
|
||||
# Copy data, ensuring to create the timestamp_ms
|
||||
logging.info(f" -> Migrating data for '{table_name}'...")
|
||||
old_df = pd.read_sql(f'SELECT * FROM "{table_name}_old"', conn, parse_dates=['datetime_utc'])
|
||||
if not old_df.empty:
|
||||
old_df['timestamp_ms'] = (old_df['datetime_utc'].astype('int64') // 10**6)
|
||||
# Keep only unique timestamps, preserving the last entry
|
||||
old_df.drop_duplicates(subset=['timestamp_ms'], keep='last', inplace=True)
|
||||
old_df.to_sql(table_name, conn, if_exists='append', index=False)
|
||||
logging.info(f" -> Data migration complete.")
|
||||
conn.execute(f'DROP TABLE "{table_name}_old"')
|
||||
conn.commit()
|
||||
logging.info(f"Successfully migrated schema for '{table_name}'.")
|
||||
except Exception as e:
|
||||
logging.error(f"FATAL: Migration for '{table_name}' failed: {e}. Please delete 'market_data.db' and restart.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
self._create_resampled_table(conn, table_name)
|
||||
logging.info("All resampled table schemas verified.")
|
||||
|
||||
def _create_resampled_table(self, conn, table_name):
|
||||
"""Creates a new resampled table with the correct schema."""
|
||||
# --- FIX: Set PRIMARY KEY on timestamp_ms for performance and uniqueness ---
|
||||
conn.execute(f'''
|
||||
CREATE TABLE "{table_name}" (
|
||||
datetime_utc TEXT,
|
||||
timestamp_ms INTEGER PRIMARY KEY,
|
||||
open REAL,
|
||||
high REAL,
|
||||
low REAL,
|
||||
close REAL,
|
||||
volume REAL,
|
||||
number_of_trades INTEGER
|
||||
)
|
||||
''')
|
||||
conn = db.get_connection()
|
||||
for coin in self.coins_to_process:
|
||||
for tf_name in self.timeframes.keys():
|
||||
table_name = db.sanitize_table_name(coin, tf_name)
|
||||
db.create_candle_table(conn, table_name)
|
||||
conn.close()
|
||||
logging.info("All resampled table schemas verified.")
|
||||
|
||||
def _load_existing_status(self) -> dict:
|
||||
"""Loads the existing status file if it exists, otherwise returns an empty dict."""
|
||||
@ -116,13 +74,8 @@ class Resampler:
|
||||
logging.warning("No timeframes to process after filtering. Exiting job.")
|
||||
return
|
||||
|
||||
if not os.path.exists(self.db_path):
|
||||
logging.error(f"Database file '{self.db_path}' not found.")
|
||||
return
|
||||
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
conn.execute("PRAGMA journal_mode=WAL;")
|
||||
|
||||
conn = db.get_connection()
|
||||
try:
|
||||
logging.debug(f"Processing {len(self.coins_to_process)} coins...")
|
||||
|
||||
for coin in self.coins_to_process:
|
||||
@ -130,8 +83,8 @@ class Resampler:
|
||||
|
||||
try:
|
||||
for tf_name, tf_code in self.timeframes.items():
|
||||
target_table_name = f"{coin}_{tf_name}"
|
||||
source_table_name = f"{coin}_1m"
|
||||
target_table_name = db.sanitize_table_name(coin, tf_name)
|
||||
source_table_name = db.sanitize_table_name(coin, "1m")
|
||||
logging.debug(f" Updating {tf_name} table...")
|
||||
|
||||
last_timestamp_ms = self._get_last_timestamp(conn, target_table_name)
|
||||
@ -139,7 +92,7 @@ class Resampler:
|
||||
query = f'SELECT * FROM "{source_table_name}"'
|
||||
params = ()
|
||||
if last_timestamp_ms:
|
||||
query += ' WHERE timestamp_ms >= ?'
|
||||
query += ' WHERE timestamp_ms >= %s'
|
||||
# Go back one interval to rebuild the last (potentially partial) candle
|
||||
try:
|
||||
interval_delta_ms = pd.to_timedelta(tf_code).total_seconds() * 1000
|
||||
@ -170,12 +123,7 @@ class Resampler:
|
||||
row['volume'], row['number_of_trades']
|
||||
))
|
||||
|
||||
cursor = conn.cursor()
|
||||
cursor.executemany(f'''
|
||||
INSERT OR REPLACE INTO "{target_table_name}" (datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
''', records_to_upsert)
|
||||
conn.commit()
|
||||
db.upsert_candles(conn, target_table_name, records_to_upsert)
|
||||
|
||||
logging.debug(f" -> Upserted {len(resampled_df)} candles into '{target_table_name}'.")
|
||||
|
||||
@ -188,6 +136,8 @@ class Resampler:
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to process coin '{coin}': {e}")
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
self._log_summary()
|
||||
self._save_status()
|
||||
|
||||
Reference in New Issue
Block a user