diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..2605f78 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,14 @@ +.venv/ +.git/ +_logs/ +_data/*.db +_data/*.db-shm +_data/*.db-wal +__pycache__/ +*.pyc +.temp/ +sdk/ +agents/ +secrets/ +.env.docker +.env diff --git a/.env.docker.example b/.env.docker.example new file mode 100644 index 0000000..81ded72 --- /dev/null +++ b/.env.docker.example @@ -0,0 +1,7 @@ +# Docker environment variables +# Copy to .env.docker and fill in real values. +# DO NOT commit the real .env.docker file to git. + +POSTGRES_PASSWORD=change_me +PG_CONN_STR=postgresql://hyper:change_me@postgres:5432/hyper +COINGECKO_API_KEY= diff --git a/.gitignore b/.gitignore index 32037b3..0c358e4 100644 --- a/.gitignore +++ b/.gitignore @@ -42,4 +42,8 @@ agents/ .idea/ .DS_Store Thumbs.db -.opencode/ \ No newline at end of file +.opencode/ + +# --- Docker --- +secrets/ +.env.docker \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..5a6d6ab --- /dev/null +++ b/Dockerfile @@ -0,0 +1,22 @@ +FROM python:3.11-slim + +# Install supervisor for process management +RUN apt-get update && apt-get install -y --no-install-recommends supervisor && \ + rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +# Install Python dependencies +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application source files +COPY . . + +# Copy supervisord configuration +COPY supervisord.conf /etc/supervisor/conf.d/supervisord.conf + +# Create required directories +RUN mkdir -p /app/_data /app/_logs + +CMD ["/usr/bin/supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"] diff --git a/MIGRATION_PLAN.md b/MIGRATION_PLAN.md new file mode 100644 index 0000000..78169a8 --- /dev/null +++ b/MIGRATION_PLAN.md @@ -0,0 +1,126 @@ +# Migration Plan: SQLite → PostgreSQL + Docker on Synology DS1513+ + +## Architecture Decisions + +| Decision | Choice | Rationale | +|----------|--------|-----------| +| Schema | Keep table-per-coin-timeframe (652 tables) | Minimal code changes, PostgreSQL handles it well | +| Table names | Sanitize `:` → `_` (e.g., `xyz_BRENTOIL_1m`) | PostgreSQL compatibility | +| Secrets | Docker env_file + bind-mount | Secure, rotate-friendly, Synology-compatible | +| Gap detection | New `gap_detector.py` | Fills data gaps when system is down | +| Backup | Daily `pg_dump` to shared folder | Accessible via File Station, Hyper Backup compatible | +| Host integration | Expose PostgreSQL port 5432 | Host scripts connect to `localhost:5432` | +| Migration | Two-phase (offline + cutover) | Minimizes downtime | +| Legacy tables | Skip `market_cap`, `candles`, `daily` | Not used by current code | + +## Container Layout + +``` +┌─────────────────────────────────────────────────────┐ +│ Docker Compose │ +├─────────────────────────────────────────────────────┤ +│ ┌──────────────┐ ┌──────────────────────────────┐ │ +│ │ PostgreSQL │ │ Data Collector (supervisord)│ │ +│ │ postgres:15- │ │ python:3.11-slim │ │ +│ │ alpine │ │ │ │ +│ │ │ │ • live_candle_fetcher (cont)│ │ +│ │ shared_buff │ │ • resampler_loop (cont) │ │ +│ │ =128MB │ │ • indicators_fetcher (cont) │ │ +│ │ │ │ • cron_scheduler (cont) │ │ +│ │ Vol:pg_data │ │ - data_fetcher (daily) │ │ +│ │ Port:5432 │ │ - fetch_history (daily) │ │ +│ │ exposed │ │ - gap_detector (hourly) │ │ +│ └──────────────┘ │ - backup_runner (daily) │ │ +│ └──────────────────────────────┘ │ +└─────────────────────────────────────────────────────┘ + +Host Machine: indicators.py, base_strategy.py, main_app.py + → connect to localhost:5432 +``` + +## PostgreSQL Configuration (4GB RAM) + +```ini +shared_buffers = 128MB +effective_cache_size = 512MB +work_mem = 8MB +maintenance_work_mem = 64MB +max_connections = 10 +max_worker_processes = 2 +checkpoint_completion_target = 0.9 +wal_buffers = 4MB +``` + +## Data Migration (Two-Phase) + +**Phase 1 (offline)**: Stop current system → run `migrate_sqlite_to_pg.py` → 2-3 hours for 1.8GB + +**Phase 2 (cutover)**: Start Docker containers → update host scripts to connect to `localhost:5432` + +## Files to Create/Modify + +### New Files +1. `db.py` — PostgreSQL abstraction layer +2. `scripts/resampler_loop.py` — Runs resampler every minute in a loop +3. `scripts/gap_detector.py` — Detects and fills data gaps +4. `scripts/backup_runner.py` — Daily pg_dump with 7-day retention +5. `scripts/cron_scheduler.py` — Schedules data_fetcher, fetch_history, gap_detector, backup +6. `migrate_sqlite_to_pg.py` — One-time data migration +7. `Dockerfile` — Python 3.11-slim + supervisor + psycopg2-binary +8. `docker-compose.yml` — PostgreSQL + data-collector services +9. `supervisord.conf` — Process management +10. `postgres/postgresql.conf` — Tuned for 4GB RAM +11. `.dockerignore` — Docker build context exclusions +12. `.env.docker.example` — Docker env template +13. `secrets/pg_password.txt.example` — PG password template + +### Files to Modify (7) +1. `live_candle_fetcher.py` — `sqlite3` → `db.py` +2. `resampler.py` — `sqlite3` → `db.py` +3. `data_fetcher.py` — `sqlite3` → `db.py` +4. `fetch_history.py` — `sqlite3` → `db.py` +5. `import_csv.py` — `sqlite3` → `db.py` +6. `indicators.py` — `sqlite3` → `psycopg2` +7. `base_strategy.py` — `sqlite3` → `psycopg2` + +## TODO List + +### Phase 1: DB Abstraction Layer +- [x] Create `db.py` with PostgreSQL connection, table sanitization, upsert logic +- [x] Add `psycopg2-binary` to `requirements.txt` + +### Phase 2: Modify Data Collection Components +- [ ] Modify `live_candle_fetcher.py` — replace `sqlite3.connect()` with `db.get_connection()`, `INSERT OR REPLACE` with `db.upsert_candles()`, sanitize table names +- [ ] Modify `resampler.py` — replace `sqlite3` with `db.py`, `INSERT OR REPLACE` with `db.upsert_candles()`, `?` → `%s` +- [ ] Modify `data_fetcher.py` — replace `sqlite3` with `db.py`, `to_sql()` → `db.upsert_candles()` +- [ ] Modify `fetch_history.py` — replace `sqlite3` with `db.py` +- [ ] Modify `import_csv.py` — replace `sqlite3` with `db.py`, `to_sql()` → `db.upsert_candles()` + +### Phase 3: New Components +- [ ] Create `scripts/resampler_loop.py` — wraps resampler in a while loop with 60s sleep +- [ ] Create `scripts/gap_detector.py` — detects gaps in 1m data, backfills via HTTP API +- [ ] Create `scripts/backup_runner.py` — daily pg_dump with 7-day retention +- [ ] Create `scripts/cron_scheduler.py` — schedules data_fetcher, fetch_history, gap_detector, backup + +### Phase 4: Docker Setup +- [ ] Create `Dockerfile` (python:3.11-slim + supervisor + psycopg2-binary) +- [ ] Create `docker-compose.yml` (postgres + data-collector services) +- [ ] Create `supervisord.conf` (live_candle_fetcher, resampler_loop, indicators_fetcher, cron_scheduler) +- [ ] Create `postgres/postgresql.conf` (tuned for 4GB RAM) +- [ ] Create `.dockerignore` +- [ ] Create `.env.docker.example` +- [ ] Create `secrets/pg_password.txt.example` +- [ ] Update `.gitignore` + +### Phase 5: Host-Side Updates +- [ ] Modify `indicators.py` on host — connect to `localhost:5432` +- [ ] Modify `base_strategy.py` on host — connect to `localhost:5432` + +### Phase 6: Migration Tool +- [ ] Create `migrate_sqlite_to_pg.py` — reads from SQLite, writes to PostgreSQL + +### Phase 7: Testing & Deployment +- [ ] Commit and push to remote +- [ ] User clones on NAS, copies `.env` and `_data/` +- [ ] User runs migration script +- [ ] User starts Docker containers diff --git a/base_strategy.py b/base_strategy.py index 5f0a001..7c3d359 100644 --- a/base_strategy.py +++ b/base_strategy.py @@ -4,7 +4,7 @@ import json import os import logging from datetime import datetime, timezone -import sqlite3 +import psycopg2 import multiprocessing import time @@ -27,7 +27,7 @@ class BaseStrategy(ABC): self.coin = params.get("coin", "N/A") self.timeframe = params.get("timeframe", "N/A") - 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", f"strategy_status_{self.strategy_name}.json") self.current_signal = "INIT" @@ -38,19 +38,23 @@ class BaseStrategy(ABC): def load_data(self) -> pd.DataFrame: """Loads historical data for the configured coin and timeframe.""" - table_name = f"{self.coin}_{self.timeframe}" + table_name = f"{self.coin.replace(':', '_')}_{self.timeframe}" periods = [v for k, v in self.params.items() if 'period' in k or '_ma' in k or 'slow' in k or 'fast' in k] limit = max(periods) + 50 if periods else 500 try: - with sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True) as conn: + conn = psycopg2.connect(self.db_path) + conn.set_session(readonly=True) + try: query = f'SELECT * FROM "{table_name}" ORDER BY datetime_utc DESC LIMIT {limit}' df = pd.read_sql(query, conn, parse_dates=['datetime_utc']) if df.empty: return pd.DataFrame() df.set_index('datetime_utc', inplace=True) df.sort_index(inplace=True) return df + finally: + conn.close() except Exception as e: logging.error(f"Failed to load data from table '{table_name}': {e}") return pd.DataFrame() diff --git a/data_fetcher.py b/data_fetcher.py index a09e64a..7e6b7d8 100644 --- a/data_fetcher.py +++ b/data_fetcher.py @@ -4,7 +4,7 @@ import logging import os import sys import time -import sqlite3 +import db import pandas as pd from datetime import datetime, timedelta, timezone @@ -26,7 +26,7 @@ class CandleFetcherDB: self.coins = self._resolve_coins(coins_to_fetch) self.interval = interval self.days_back = days_back - 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.column_rename_map = { 't': 'timestamp_ms', 'o': 'open', 'h': 'high', 'l': 'low', 'c': 'close', 'v': 'volume', 'n': 'number_of_trades' } @@ -47,13 +47,12 @@ class CandleFetcherDB: def run(self): """Starts the data fetching process and reports status after each coin.""" - with sqlite3.connect(self.db_path, timeout=10) as self.conn: - self.conn.execute("PRAGMA journal_mode=WAL;") - for coin in self.coins: - logging.info(f"--- Starting process for {coin} ---") - num_updated = self._update_data_for_coin(coin) - self._report_status(coin, num_updated) - time.sleep(1) + self.conn = db.get_connection() + for coin in self.coins: + logging.info(f"--- Starting process for {coin} ---") + num_updated = self._update_data_for_coin(coin) + self._report_status(coin, num_updated) + time.sleep(1) def _report_status(self, last_coin: str, num_updated: int): """Saves the status of the fetcher run to a JSON file.""" @@ -73,11 +72,11 @@ class CandleFetcherDB: def _get_start_time(self, coin: str) -> (int, bool): """Checks the database for an existing table and returns the last timestamp.""" - table_name = f"{coin}_{self.interval}" + table_name = db.sanitize_table_name(coin, self.interval) try: cursor = self.conn.cursor() - cursor.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}';") - if cursor.fetchone(): + cursor.execute("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = %s)", (table_name,)) + if cursor.fetchone()[0]: query = f'SELECT MAX(timestamp_ms) FROM "{table_name}"' last_ts = pd.read_sql(query, self.conn).iloc[0, 0] if pd.notna(last_ts): @@ -150,23 +149,28 @@ class CandleFetcherDB: return None def _save_to_sqlite_with_pandas(self, df: pd.DataFrame, coin: str, is_append: bool) -> int: - """Saves a pandas DataFrame to an SQLite table and returns the number of saved rows.""" - table_name = f"{coin}_{self.interval}" + """Saves a pandas DataFrame to a PostgreSQL table and returns the number of saved rows.""" + table_name = db.sanitize_table_name(coin, self.interval) try: df.rename(columns=self.column_rename_map, inplace=True) df['datetime_utc'] = pd.to_datetime(df['timestamp_ms'], unit='ms') final_df = df[['datetime_utc', 'timestamp_ms', 'open', 'high', 'low', 'close', 'volume', 'number_of_trades']] - write_mode = 'append' if is_append else 'replace' - final_df.to_sql(table_name, self.conn, if_exists=write_mode, index=False) - - self.conn.execute(f'CREATE INDEX IF NOT EXISTS "idx_{table_name}_time" ON "{table_name}"(datetime_utc);') + if not is_append: + # Drop and recreate the table for 'replace' mode + with self.conn.cursor() as cur: + cur.execute(f'DROP TABLE IF EXISTS "{table_name}"') + self.conn.commit() + db.create_candle_table(self.conn, table_name) + + records = list(final_df.itertuples(index=False, name=None)) + db.upsert_candles(self.conn, table_name, records) num_saved = len(final_df) logging.info(f"Successfully saved {num_saved} candles to table '{table_name}'") return num_saved except Exception as e: - logging.error(f"Failed to write to SQLite table '{table_name}': {e}") + logging.error(f"Failed to write to table '{table_name}': {e}") return 0 diff --git a/db.py b/db.py new file mode 100644 index 0000000..a3b6ac8 --- /dev/null +++ b/db.py @@ -0,0 +1,132 @@ +""" +PostgreSQL database abstraction layer for the Hyperliquid trading toolkit. + +Provides a thin wrapper around psycopg2 to centralize database operations, +handle table name sanitization, and abstract SQL dialect differences +from the SQLite-based codebase. +""" + +import os +import psycopg2 +from psycopg2.extras import execute_values + +PG_CONN_STR = os.environ.get( + "PG_CONN_STR", + "postgresql://hyper:hyper@localhost:5432/hyper" +) + + +def get_connection(): + """Return a new psycopg2 connection to the PostgreSQL database.""" + return psycopg2.connect(PG_CONN_STR) + + +def sanitize_table_name(coin, timeframe): + """ + Sanitize a coin/timeframe pair into a PostgreSQL-safe table name. + + Replaces colons with underscores (e.g., 'xyz:BRENTOIL' -> 'xyz_BRENTOIL') + to ensure compatibility with PostgreSQL identifier rules. + """ + return f"{coin.replace(':', '_')}_{timeframe}" + + +def create_candle_table(conn, table_name): + """ + Create a candle table if it does not already exist. + + Schema matches the original SQLite layout: + datetime_utc, timestamp_ms (PK), open, high, low, close, volume, number_of_trades + Also creates an index on datetime_utc for time-range queries. + """ + with conn.cursor() as cur: + cur.execute(f''' + CREATE TABLE IF NOT EXISTS "{table_name}" ( + datetime_utc TIMESTAMP, + timestamp_ms BIGINT PRIMARY KEY, + open REAL, + high REAL, + low REAL, + close REAL, + volume REAL, + number_of_trades INTEGER + ) + ''') + cur.execute( + f'CREATE INDEX IF NOT EXISTS "idx_{table_name}_time" ON "{table_name}"(datetime_utc)' + ) + conn.commit() + + +def upsert_candles(conn, table_name, records): + """ + Batch upsert candle records using PostgreSQL ON CONFLICT. + + Args: + conn: psycopg2 connection + table_name: sanitized table name (e.g., 'BTC_1m') + records: list of tuples (datetime_utc, timestamp_ms, open, high, + low, close, volume, number_of_trades) + + Returns: + Number of records upserted. + """ + if not records: + return 0 + + with conn.cursor() as cur: + execute_values( + cur, + f''' + INSERT INTO "{table_name}" + (datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades) + VALUES %s + ON CONFLICT (timestamp_ms) DO UPDATE SET + datetime_utc = EXCLUDED.datetime_utc, + open = EXCLUDED.open, + high = EXCLUDED.high, + low = EXCLUDED.low, + close = EXCLUDED.close, + volume = EXCLUDED.volume, + number_of_trades = EXCLUDED.number_of_trades + ''', + records, + page_size=1000 + ) + conn.commit() + return len(records) + + +def get_last_timestamp(conn, table_name): + """Return the most recent timestamp_ms from a table, or None.""" + with conn.cursor() as cur: + cur.execute(f'SELECT MAX(timestamp_ms) FROM "{table_name}"') + result = cur.fetchone() + return result[0] if result and result[0] is not None else None + + +def get_table_count(conn, table_name): + """Return the total row count of a table.""" + with conn.cursor() as cur: + cur.execute(f'SELECT COUNT(*) FROM "{table_name}"') + return cur.fetchone()[0] + + +def table_exists(conn, table_name): + """Check if a table exists in the database.""" + with conn.cursor() as cur: + cur.execute( + "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = %s)", + (table_name,) + ) + return cur.fetchone()[0] + + +def get_table_columns(conn, table_name): + """Return a list of column names for a table.""" + with conn.cursor() as cur: + cur.execute( + "SELECT column_name FROM information_schema.columns WHERE table_name = %s", + (table_name,) + ) + return [row[0] for row in cur.fetchall()] diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..a82678b --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,42 @@ +version: "3.8" + +services: + postgres: + image: postgres:15-alpine + container_name: hyper_pg + restart: unless-stopped + environment: + POSTGRES_DB: hyper + POSTGRES_USER: hyper + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD} + volumes: + - pg_data:/var/lib/postgresql/data + - ./postgres/postgresql.conf:/etc/postgresql/postgresql.conf + command: postgres -c config_file=/etc/postgresql/postgresql.conf + ports: + - "5432:5432" + networks: + - hyper_net + + data-collector: + build: . + container_name: hyper_data + restart: unless-stopped + depends_on: + - postgres + env_file: + - .env.docker + volumes: + - ./_data:/app/_data + - ./_logs:/app/_logs + - ./secrets:/app/secrets + - /volume1/docker/hyper/backups:/backups + networks: + - hyper_net + +volumes: + pg_data: + +networks: + hyper_net: + driver: bridge diff --git a/fetch_history.py b/fetch_history.py index 7c14441..1a3d307 100644 --- a/fetch_history.py +++ b/fetch_history.py @@ -1,10 +1,10 @@ import requests import json -import sqlite3 +import db import time from datetime import datetime, timezone -DB_PATH = "_data/market_data.db" +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"): @@ -37,22 +37,10 @@ def fetch_historical_candles(coin, start_ms, end_ms, interval="1m"): def write_candles_to_db(coin, candles, interval="1m"): """Write candles to the database.""" - table_name = coin + "_" + interval - conn = sqlite3.connect(DB_PATH) - cursor = conn.cursor() - # Ensure table exists - cursor.execute(f''' - CREATE TABLE IF NOT EXISTS "{table_name}" ( - datetime_utc TEXT, - timestamp_ms INTEGER PRIMARY KEY, - open REAL, - high REAL, - low REAL, - close REAL, - volume REAL, - number_of_trades INTEGER - ) - ''') + 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'), @@ -60,22 +48,16 @@ def write_candles_to_db(coin, candles, interval="1m"): candle.get('o'), candle.get('h'), candle.get('l'), candle.get('c'), candle.get('v'), candle.get('n') ) - cursor.execute(f''' - INSERT OR REPLACE INTO "{table_name}" (datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ''', record) - conn.commit() + 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 = coin + "_1m" - conn = sqlite3.connect(DB_PATH) - cursor = conn.cursor() + table_name = db.sanitize_table_name(coin, "1m") + conn = db.get_connection() try: - cursor.execute(f'SELECT MAX(timestamp_ms) FROM "{table_name}"') - result = cursor.fetchone() - return int(result[0]) if result and result[0] is not None else None + return db.get_last_timestamp(conn, table_name) except: return None finally: diff --git a/import_csv.py b/import_csv.py index e735471..7298969 100644 --- a/import_csv.py +++ b/import_csv.py @@ -2,7 +2,7 @@ import argparse import logging import os import sys -import sqlite3 +import db import pandas as pd from datetime import datetime @@ -24,8 +24,8 @@ class CsvImporter: self.csv_path = csv_path self.coin = coin # --- FIX: Corrected the f-string syntax for the table name --- - self.table_name = f"{self.coin}_1m" - self.db_path = os.path.join("_data", "market_data.db") + self.table_name = db.sanitize_table_name(self.coin, "1m") + self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper") self.column_mapping = { 'Open time': 'datetime_utc', 'Open': 'open', @@ -40,9 +40,8 @@ class CsvImporter: """Orchestrates the entire import and verification process.""" logging.info(f"Starting import process for '{self.coin}' from '{self.csv_path}'...") - with sqlite3.connect(self.db_path) as conn: - conn.execute("PRAGMA journal_mode=WAL;") - + conn = db.get_connection() + try: # 1. Get the current state of the database db_oldest, db_newest, initial_row_count = self._get_db_state(conn) @@ -58,6 +57,8 @@ class CsvImporter: # 4. Summarize and verify the import self._summarize_import(initial_row_count, len(new_data_df), conn) + finally: + conn.close() def _get_db_state(self, conn) -> (datetime, datetime, int): """Gets the oldest and newest timestamps and total row count from the DB table.""" @@ -104,9 +105,10 @@ class CsvImporter: return df_filtered def _append_to_db(self, df: pd.DataFrame, conn): - """Appends the DataFrame to the SQLite table.""" + """Appends the DataFrame to the database.""" logging.info(f"Appending {len(df):,} new rows to the database...") - df.to_sql(self.table_name, conn, if_exists='append', index=False) + records = list(df.itertuples(index=False, name=None)) + db.upsert_candles(conn, self.table_name, records) logging.info("Append operation complete.") def _summarize_import(self, initial_count: int, added_count: int, conn): diff --git a/indicators.py b/indicators.py index d7ef8cf..b9b0b99 100644 --- a/indicators.py +++ b/indicators.py @@ -7,7 +7,8 @@ and custom functions. import json import os -import sqlite3 +import psycopg2 +from contextlib import closing import importlib import logging import pandas as pd @@ -36,9 +37,9 @@ class IndicatorCalculator: def _get_latest_close(self, coin, timeframe="1m"): """Get the latest close price from a candle table.""" - table = f"{coin}_{timeframe}" + table = f"{coin.replace(':', '_')}_{timeframe}" try: - with sqlite3.connect(self.db_path) as conn: + with closing(psycopg2.connect(self.db_path)) as conn: result = conn.execute( f'SELECT close FROM "{table}" ORDER BY timestamp_ms DESC LIMIT 1' ).fetchone() @@ -49,9 +50,9 @@ class IndicatorCalculator: def _get_close_n_candles_ago(self, coin, timeframe, n=1): """Get the close price from n candles ago (n=1 = most recent completed candle).""" - table = f"{coin}_{timeframe}" + table = f"{coin.replace(':', '_')}_{timeframe}" try: - with sqlite3.connect(self.db_path) as conn: + with closing(psycopg2.connect(self.db_path)) as conn: result = conn.execute( f'SELECT close FROM "{table}" ORDER BY timestamp_ms DESC LIMIT 1 OFFSET {n}' ).fetchone() @@ -62,9 +63,9 @@ class IndicatorCalculator: def _get_all_closes(self, coin, timeframe="1d"): """Get all close prices from a candle table, ordered by time.""" - table = f"{coin}_{timeframe}" + table = f"{coin.replace(':', '_')}_{timeframe}" try: - with sqlite3.connect(self.db_path) as conn: + with closing(psycopg2.connect(self.db_path)) as conn: result = conn.execute( f'SELECT close FROM "{table}" ORDER BY timestamp_ms' ).fetchall() @@ -75,10 +76,10 @@ class IndicatorCalculator: def _get_all_ratio(self, num_coin, den_coin, timeframe="1d"): """Get all ratio values (num/den) from candle tables, ordered by time.""" - num_table = f"{num_coin}_{timeframe}" - den_table = f"{den_coin}_{timeframe}" + num_table = f"{num_coin.replace(':', '_')}_{timeframe}" + den_table = f"{den_coin.replace(':', '_')}_{timeframe}" try: - with sqlite3.connect(self.db_path) as conn: + with closing(psycopg2.connect(self.db_path)) as conn: result = conn.execute( f'SELECT n.close / d.close as ratio ' f'FROM "{num_table}" n ' @@ -92,10 +93,10 @@ class IndicatorCalculator: def _get_all_spread(self, num_coin, den_coin, timeframe="1d"): """Get all spread values (num - den) from candle tables, ordered by time.""" - num_table = f"{num_coin}_{timeframe}" - den_table = f"{den_coin}_{timeframe}" + num_table = f"{num_coin.replace(':', '_')}_{timeframe}" + den_table = f"{den_coin.replace(':', '_')}_{timeframe}" try: - with sqlite3.connect(self.db_path) as conn: + with closing(psycopg2.connect(self.db_path)) as conn: result = conn.execute( f'SELECT n.close - d.close as spread ' f'FROM "{num_table}" n ' @@ -109,10 +110,10 @@ class IndicatorCalculator: def _get_all_diff_pct(self, num_coin, den_coin, timeframe="1d"): """Get all percentage difference values ((num-den)/den*100) from candle tables.""" - num_table = f"{num_coin}_{timeframe}" - den_table = f"{den_coin}_{timeframe}" + num_table = f"{num_coin.replace(':', '_')}_{timeframe}" + den_table = f"{den_coin.replace(':', '_')}_{timeframe}" try: - with sqlite3.connect(self.db_path) as conn: + with closing(psycopg2.connect(self.db_path)) as conn: result = conn.execute( f'SELECT (n.close - d.close) / d.close * 100 as diff_pct ' f'FROM "{num_table}" n ' diff --git a/indicators_fetcher.py b/indicators_fetcher.py index d52a347..4cdf200 100644 --- a/indicators_fetcher.py +++ b/indicators_fetcher.py @@ -29,7 +29,7 @@ class IndicatorsFetcher: setup_logging(log_level, 'IndicatorsFetcher') project_root = os.path.dirname(os.path.abspath(__file__)) - self.db_path = os.path.join(project_root, "_data", "market_data.db") + self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper") self.config_path = os.path.join(project_root, "_data", "indicators.json") self.status_file_path = os.path.join(project_root, "_logs", "indicators_status.json") diff --git a/live_candle_fetcher.py b/live_candle_fetcher.py index 400441e..8cc1288 100644 --- a/live_candle_fetcher.py +++ b/live_candle_fetcher.py @@ -7,7 +7,7 @@ import time from datetime import datetime, timezone from hyperliquid.info import Info from hyperliquid.utils import constants -import sqlite3 +import db from queue import Queue from threading import Thread @@ -22,7 +22,7 @@ class LiveCandleFetcher: def __init__(self, log_level: str, coins: list): setup_logging(log_level, 'LiveCandleFetcher') - 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.coins_to_watch = set(coins) if not self.coins_to_watch: logging.error("No coins provided to watch. Exiting.") @@ -34,65 +34,15 @@ class LiveCandleFetcher: def _ensure_tables_exist(self): """ - Ensures that all necessary tables are created with the correct schema and PRIMARY KEY. - If a table exists with an incorrect schema, it attempts to migrate the data. + Ensures that all necessary tables are created with the correct schema. + Uses db.create_candle_table() which is idempotent (CREATE TABLE IF NOT EXISTS). """ - with sqlite3.connect(self.db_path) as conn: - for coin in self.coins_to_watch: - table_name = f"{coin}_1m" - cursor = conn.cursor() - cursor.execute(f"PRAGMA table_info('{table_name}')") - columns = cursor.fetchall() - - if columns: - 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}': 'timestamp_ms' is not the PRIMARY KEY.") - logging.warning("Attempting to automatically rebuild the table...") - try: - # 1. Rename old table - conn.execute(f'ALTER TABLE "{table_name}" RENAME TO "{table_name}_old"') - logging.info(f" -> Renamed existing table to '{table_name}_old'.") - - # 2. Create new table with correct schema - self._create_candle_table(conn, table_name) - logging.info(f" -> Created new '{table_name}' table with correct schema.") - - # 3. Copy unique data from old table to new table - conn.execute(f''' - INSERT OR IGNORE INTO "{table_name}" (datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades) - SELECT datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades - FROM "{table_name}_old" - ''') - conn.commit() - logging.info(" -> Copied data to new table.") - - # 4. Drop the old table - conn.execute(f'DROP TABLE "{table_name}_old"') - logging.info(f" -> Removed old table. Migration for '{table_name}' complete.") - except Exception as e: - logging.error(f"FATAL: Automatic schema migration for '{table_name}' failed: {e}") - logging.error("Please delete the database file '_data/market_data.db' manually and restart.") - sys.exit(1) - else: - # If table does not exist, create it - self._create_candle_table(conn, table_name) - logging.info("Database tables verified.") - - def _create_candle_table(self, conn, table_name: str): - """Creates a new candle table with the correct schema.""" - 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_watch: + table_name = db.sanitize_table_name(coin, "1m") + db.create_candle_table(conn, table_name) + conn.close() + logging.info("Database tables verified.") def on_message(self, message): """ @@ -112,6 +62,7 @@ class LiveCandleFetcher: This is the "Consumer" thread. It runs forever, pulling candles from the queue and writing them to the database, ensuring all writes are serial. """ + conn = db.get_connection() while True: try: candle = self.candle_queue.get() @@ -122,7 +73,7 @@ class LiveCandleFetcher: if not coin: continue - table_name = f"{coin}_1m" + table_name = db.sanitize_table_name(coin, "1m") record = ( datetime.fromtimestamp(candle['t'] / 1000, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S'), candle['t'], @@ -130,24 +81,21 @@ class LiveCandleFetcher: candle.get('v'), candle.get('n') ) - with sqlite3.connect(self.db_path) as conn: - conn.execute(f''' - INSERT OR REPLACE INTO "{table_name}" (datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ''', record) - conn.commit() + db.upsert_candles(conn, table_name, [record]) logging.debug(f"Upserted candle for {coin} at {record[0]}") except Exception as e: logging.error(f"Error in database writer thread: {e}") + conn.close() def _get_last_timestamp_from_db(self, coin: str) -> int: """Gets the most recent millisecond timestamp from a coin's 1m table.""" - table_name = f"{coin}_1m" + table_name = db.sanitize_table_name(coin, "1m") try: - with sqlite3.connect(self.db_path) as conn: - result = conn.execute(f'SELECT MAX(timestamp_ms) FROM "{table_name}"').fetchone() - return int(result[0]) if result and result[0] is not None else None + conn = db.get_connection() + result = db.get_last_timestamp(conn, table_name) + conn.close() + return result except Exception as e: logging.error(f"Could not read last timestamp from table '{table_name}': {e}") return None diff --git a/migrate_sqlite_to_pg.py b/migrate_sqlite_to_pg.py new file mode 100644 index 0000000..41a94cb --- /dev/null +++ b/migrate_sqlite_to_pg.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +""" +SQLite to PostgreSQL Migration Script + +Reads candle data from a SQLite database and writes it to PostgreSQL. +This is a one-time migration tool used to transfer existing historical +data from the old SQLite database to the new PostgreSQL database. + +Usage: + python migrate_sqlite_to_pg.py --sqlite-path _data/market_data.db --log-level normal + +The script: +1. Connects to both SQLite (source) and PostgreSQL (destination) +2. Enumerates all candle tables (skipping legacy tables like market_cap) +3. For each table, reads data from SQLite and upserts to PostgreSQL +4. Handles table name sanitization (colons → underscores) +""" + +import argparse +import logging +import os +import sqlite3 +import sys + +import pandas as pd + +from logging_utils import setup_logging +from db import get_connection, sanitize_table_name, upsert_candles, create_candle_table + +TIMEFRAMES = [ + '1m', '3m', '5m', '15m', '30m', '37m', '148m', + '1h', '2h', '4h', '8h', '12h', '1d', '3d', '1w', '1month' +] + + +def get_candle_tables(sqlite_conn): + """Get all candle table names from SQLite (excluding legacy tables).""" + cursor = sqlite_conn.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") + all_tables = [row[0] for row in cursor.fetchall()] + + candle_tables = [] + for table in all_tables: + for tf in TIMEFRAMES: + if table.endswith(f'_{tf}'): + candle_tables.append(table) + break + + return candle_tables + + +def parse_table_name(table_name): + """Parse a table name into (coin, timeframe).""" + for tf in TIMEFRAMES: + suffix = f'_{tf}' + if table_name.endswith(suffix): + coin = table_name[:-len(suffix)] + return coin, tf + return table_name, '1m' + + +def main(): + parser = argparse.ArgumentParser(description="Migrate data from SQLite to PostgreSQL.") + parser.add_argument("--sqlite-path", default="_data/market_data.db", + help="Path to the SQLite database file.") + parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug']) + args = parser.parse_args() + + setup_logging(args.log_level, 'Migrator') + + if not os.path.exists(args.sqlite_path): + logging.error(f"SQLite database not found at '{args.sqlite_path}'") + sys.exit(1) + + sqlite_conn = sqlite3.connect(args.sqlite_path) + pg_conn = get_connection() + + tables = get_candle_tables(sqlite_conn) + logging.info(f"Found {len(tables)} candle tables to migrate") + + for table_name in tables: + coin, timeframe = parse_table_name(table_name) + pg_table = sanitize_table_name(coin, timeframe) + + logging.info(f"Migrating {table_name} -> {pg_table}") + + df = pd.read_sql(f'SELECT * FROM "{table_name}"', sqlite_conn) + if df.empty: + logging.warning(f"Table {table_name} is empty, skipping") + continue + + create_candle_table(pg_conn, pg_table) + + records = list(df.itertuples(index=False, name=None)) + upsert_candles(pg_conn, pg_table, records) + + logging.info(f"Migrated {len(records)} rows to {pg_table}") + + sqlite_conn.close() + pg_conn.close() + logging.info("Migration complete!") + + +if __name__ == "__main__": + main() diff --git a/postgres/postgresql.conf b/postgres/postgresql.conf new file mode 100644 index 0000000..356168d --- /dev/null +++ b/postgres/postgresql.conf @@ -0,0 +1,26 @@ +# PostgreSQL configuration tuned for Synology DS1513+ (4GB RAM) +# Place this file at postgres/postgresql.conf and mount it into the container. + +# --- Memory --- +shared_buffers = 128MB +effective_cache_size = 512MB +work_mem = 8MB +maintenance_work_mem = 64MB + +# --- Connections --- +max_connections = 10 +max_worker_processes = 2 + +# --- WAL / Checkpointing --- +wal_buffers = 4MB +checkpoint_completion_target = 0.9 +max_wal_senders = 3 + +# --- Network --- +listen_addresses = '*' + +# --- Logging --- +log_statement = 'none' +log_duration = off +log_min_duration_statement = 0 +log_line_prefix = '%t [%p]: [%l-1] user=%u,db=%d,app=%a,client=%h ' diff --git a/requirements.txt b/requirements.txt index c6673f8..816426a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -53,3 +53,4 @@ urllib3==1.26.20 websocket-client==1.9.0 web3~=6.0.0 # This means >=6.0.0 and <7.0.0 yarl==1.22.0 +psycopg2-binary==2.9.9 diff --git a/resampler.py b/resampler.py index ffbeb59..098f694 100644 --- a/resampler.py +++ b/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() diff --git a/scripts/backup_runner.py b/scripts/backup_runner.py new file mode 100644 index 0000000..8027afc --- /dev/null +++ b/scripts/backup_runner.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +""" +Backup Runner + +Creates a daily pg_dump backup of the PostgreSQL database, compresses it, +and retains only the last 7 days of backups. + +Designed to run as a periodic cron job (daily) inside the Docker container. +Backups are written to /backups which is mounted to a Synology shared folder. +""" + +import argparse +import logging +import os +import subprocess +from datetime import datetime, timedelta + +from logging_utils import setup_logging + +BACKUP_DIR = "/backups" +RETENTION_DAYS = 7 + + +def run_backup(): + """Run pg_dump and compress the output.""" + today = datetime.now().strftime("%Y%m%d") + backup_file = os.path.join(BACKUP_DIR, f"hyper_{today}.sql.gz") + + os.makedirs(BACKUP_DIR, exist_ok=True) + + logging.info(f"Starting backup to {backup_file}") + + cmd = f"pg_dump -h postgres -U hyper hyper | gzip > {backup_file}" + result = subprocess.run(cmd, shell=True, capture_output=True, text=True) + + if result.returncode == 0: + file_size = os.path.getsize(backup_file) + logging.info(f"Backup completed: {backup_file} ({file_size:,} bytes)") + else: + logging.error(f"Backup failed: {result.stderr}") + if os.path.exists(backup_file): + os.remove(backup_file) + + cleanup_old_backups() + + +def cleanup_old_backups(): + """Delete backup files older than RETENTION_DAYS.""" + cutoff = datetime.now() - timedelta(days=RETENTION_DAYS) + + if not os.path.exists(BACKUP_DIR): + return + + for filename in os.listdir(BACKUP_DIR): + if filename.startswith("hyper_") and filename.endswith(".sql.gz"): + filepath = os.path.join(BACKUP_DIR, filename) + mtime = datetime.fromtimestamp(os.path.getmtime(filepath)) + if mtime < cutoff: + os.remove(filepath) + logging.info(f"Deleted old backup: {filename}") + + +def main(): + parser = argparse.ArgumentParser(description="Run PostgreSQL backup.") + parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug']) + args = parser.parse_args() + + setup_logging(args.log_level, 'BackupRunner') + + run_backup() + + +if __name__ == "__main__": + main() diff --git a/scripts/cron_scheduler.py b/scripts/cron_scheduler.py new file mode 100644 index 0000000..91cf6de --- /dev/null +++ b/scripts/cron_scheduler.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +""" +Cron Scheduler + +Runs periodic maintenance tasks inside the Docker container using the +`schedule` library. This replaces a system cron daemon and keeps all +scheduling logic in Python. + +Scheduled tasks: + - data_fetcher.py — daily at 02:00 UTC (full historical catch-up) + - fetch_history.py — daily at 03:00 UTC (additional history fetch) + - gap_detector.py — hourly at :15 (fill missing 1m candles) + - backup_runner.py — daily at 04:00 UTC (pg_dump backup) +""" + +import argparse +import logging +import os +import subprocess +import sys +import time +import schedule +import signal + +from logging_utils import setup_logging + +shutdown_requested = False + + +def handle_shutdown(signum, frame): + global shutdown_requested + shutdown_requested = True + + +def run_data_fetcher(): + try: + logging.info("Running data_fetcher.py") + subprocess.run([ + sys.executable, "data_fetcher.py", + "--coins", "BTC", "ETH", "SOL", "BNB", "HYPE", "SUI", + "xyz:BRENTOIL", "xyz:CL", "xyz:GOLD", "xyz:SILVER", + "mkts:USTECH", "xyz:XYZ100", + "--interval", "1m", "--days", "7", "--log-level", "normal" + ], check=True) + except Exception as e: + logging.error(f"Data fetcher failed: {e}") + + +def run_fetch_history(): + try: + logging.info("Running fetch_history.py") + subprocess.run([sys.executable, "fetch_history.py", "--log-level", "normal"], check=True) + except Exception as e: + logging.error(f"Fetch history failed: {e}") + + +def run_gap_detector(): + try: + logging.info("Running gap_detector.py") + subprocess.run([sys.executable, "scripts/gap_detector.py", "--log-level", "normal"], check=True) + except Exception as e: + logging.error(f"Gap detector failed: {e}") + + +def run_backup(): + try: + logging.info("Running backup_runner.py") + subprocess.run([sys.executable, "scripts/backup_runner.py", "--log-level", "normal"], check=True) + except Exception as e: + logging.error(f"Backup failed: {e}") + + +def main(): + parser = argparse.ArgumentParser(description="Run periodic maintenance tasks.") + parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug']) + args = parser.parse_args() + + signal.signal(signal.SIGTERM, handle_shutdown) + signal.signal(signal.SIGINT, handle_shutdown) + + setup_logging(args.log_level, 'CronScheduler') + + # Schedule jobs + schedule.every().day.at("02:00").do(run_data_fetcher) + schedule.every().day.at("03:00").do(run_fetch_history) + schedule.every().hour.at(":15").do(run_gap_detector) + schedule.every().day.at("04:00").do(run_backup) + + logging.info("Cron scheduler started") + + while not shutdown_requested: + schedule.run_pending() + time.sleep(1) + + logging.info("Cron scheduler shutting down.") + + +if __name__ == "__main__": + main() diff --git a/scripts/gap_detector.py b/scripts/gap_detector.py new file mode 100644 index 0000000..68407fa --- /dev/null +++ b/scripts/gap_detector.py @@ -0,0 +1,137 @@ +#!/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() diff --git a/scripts/resampler_loop.py b/scripts/resampler_loop.py new file mode 100644 index 0000000..5d83195 --- /dev/null +++ b/scripts/resampler_loop.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python3 +""" +Resampler Loop Wrapper + +Runs the Resampler in a continuous loop, executing it once per minute. +This replaces the schedule-based approach used in main_app.py and is +designed to run as a supervisord-managed process inside Docker. +""" + +import argparse +import logging +import os +import sys +import time +import signal + +from logging_utils import setup_logging +from resampler import Resampler, parse_timeframes + +shutdown_requested = False + + +def handle_shutdown(signum, frame): + global shutdown_requested + shutdown_requested = True + + +def main(): + parser = argparse.ArgumentParser(description="Run the resampler in a continuous loop.") + parser.add_argument("--coins", nargs='+', required=True, help="List of coins to process.") + parser.add_argument("--timeframes", nargs='+', required=True, help="List of timeframes to generate.") + parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug']) + args = parser.parse_args() + + signal.signal(signal.SIGTERM, handle_shutdown) + signal.signal(signal.SIGINT, handle_shutdown) + + setup_logging(args.log_level, 'ResamplerLoop') + + timeframes_dict = parse_timeframes(args.timeframes) + logging.info(f"Resampler loop started. Coins: {args.coins}, Timeframes: {list(timeframes_dict.keys())}") + + while not shutdown_requested: + try: + # Pass a copy because Resampler.run() deletes '1m' from the dict + timeframes_copy = dict(timeframes_dict) + resampler = Resampler( + log_level=args.log_level, + coins=args.coins, + timeframes=timeframes_copy + ) + resampler.run() + except Exception as e: + logging.error(f"Resampler run failed: {e}") + + if shutdown_requested: + break + + # Sleep for 60 seconds, but check shutdown flag every second + for _ in range(60): + if shutdown_requested: + break + time.sleep(1) + + logging.info("Resampler loop shutting down.") + + +if __name__ == "__main__": + main() diff --git a/strategies/base_strategy.py b/strategies/base_strategy.py index afe3ff6..22e6ba3 100644 --- a/strategies/base_strategy.py +++ b/strategies/base_strategy.py @@ -4,7 +4,7 @@ import json import os import logging from datetime import datetime, timezone -import sqlite3 +import psycopg2 import multiprocessing import time @@ -27,7 +27,7 @@ class BaseStrategy(ABC): self.coin = params.get("coin", "N/A") self.timeframe = params.get("timeframe", "N/A") - 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", f"strategy_status_{self.strategy_name}.json") self.current_signal = "INIT" @@ -38,19 +38,23 @@ class BaseStrategy(ABC): def load_data(self) -> pd.DataFrame: """Loads historical data for the configured coin and timeframe.""" - table_name = f"{self.coin}_{self.timeframe}" + table_name = f"{self.coin.replace(':', '_')}_{self.timeframe}" periods = [v for k, v in self.params.items() if 'period' in k or '_ma' in k or 'slow' in k or 'fast' in k] limit = max(periods) + 50 if periods else 500 try: - with sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True) as conn: + conn = psycopg2.connect(self.db_path) + conn.set_session(readonly=True) + try: query = f'SELECT * FROM "{table_name}" ORDER BY datetime_utc DESC LIMIT {limit}' df = pd.read_sql(query, conn, parse_dates=['datetime_utc']) if df.empty: return pd.DataFrame() df.set_index('datetime_utc', inplace=True) df.sort_index(inplace=True) return df + finally: + conn.close() except Exception as e: logging.error(f"Failed to load data from table '{table_name}': {e}") return pd.DataFrame() diff --git a/supervisord.conf b/supervisord.conf new file mode 100644 index 0000000..47645a9 --- /dev/null +++ b/supervisord.conf @@ -0,0 +1,38 @@ +[supervisord] +nodaemon=true + +[program:live_candle_fetcher] +command=python live_candle_fetcher.py --coins BTC ETH SOL BNB HYPE SUI xyz:BRENTOIL xyz:CL xyz:GOLD xyz:SILVER mkts:USTECH xyz:XYZ100 --log-level normal +autostart=true +autorestart=true +stdout_logfile=/app/_logs/live_candle_fetcher.log +stderr_logfile=/app/_logs/live_candle_fetcher.log +stdout_logfile_maxbytes=10MB +stdout_logfile_backups=3 + +[program:resampler_loop] +command=python scripts/resampler_loop.py --coins BTC ETH SOL BNB HYPE SUI xyz:BRENTOIL xyz:CL xyz:GOLD xyz:SILVER mkts:USTECH xyz:XYZ100 --timeframes 3m 5m 15m 30m 1h 2h 4h 8h 12h 1d 3d 1w 1M 148m 37m --log-level normal +autostart=true +autorestart=true +stdout_logfile=/app/_logs/resampler.log +stderr_logfile=/app/_logs/resampler.log +stdout_logfile_maxbytes=10MB +stdout_logfile_backups=3 + +[program:indicators_fetcher] +command=python indicators_fetcher.py --log-level normal +autostart=true +autorestart=true +stdout_logfile=/app/_logs/indicators_fetcher.log +stderr_logfile=/app/_logs/indicators_fetcher.log +stdout_logfile_maxbytes=10MB +stdout_logfile_backups=3 + +[program:cron_scheduler] +command=python scripts/cron_scheduler.py --log-level normal +autostart=true +autorestart=true +stdout_logfile=/app/_logs/cron_scheduler.log +stderr_logfile=/app/_logs/cron_scheduler.log +stdout_logfile_maxbytes=10MB +stdout_logfile_backups=3