hyper/testing/read_candles.py

#!/usr/bin/env python3
"""
Candle Reader Test Script

Reads last 1m candles from PostgreSQL database on NAS.

Usage:
  python testing/read_candles.py                        # Last candles for 12 default symbols
  python testing/read_candles.py --all                  # Last candles for all symbols
  python testing/read_candles.py --BTC                  # Last candles for BTC only
  python testing/read_candles.py --BTC --ETH            # Last candles for BTC and ETH
  python testing/read_candles.py --xyz:GOLD           # Last candles for xyz:GOLD only
  python testing/read_candles.py --timestamp 05-08-26   # Nearest candle to Aug 5, 2026
  python testing/read_candles.py --timestamp "05-08-26 13:30"  # Nearest candle to Aug 5, 2026 13:30
  python testing/read_candles.py --BTC --timestamp 05-08-26

Environment:
  PG_HOST, PG_PORT, PG_DB, PG_USER, PG_PASSWORD, PG_TIMEOUT
  (loaded from testing/.env)
"""

import os
import sys
import argparse
from datetime import datetime

try:
    from dotenv import load_dotenv
    _TESTING_DIR = os.path.dirname(os.path.abspath(__file__))
    load_dotenv(os.path.join(_TESTING_DIR, ".env"))
except ImportError:
    pass

_PROJECT_ROOT = os.path.dirname(_TESTING_DIR)
sys.path.insert(0, _PROJECT_ROOT)

import db

PG_HOST = os.environ.get("PG_HOST", "20.20.20.20")
PG_PORT = int(os.environ.get("PG_PORT", "5433"))
PG_DB = os.environ.get("PG_DB", "hyper")
PG_USER = os.environ.get("PG_USER", "hyper")
PG_PASSWORD = os.environ.get("PG_PASSWORD", "kaqpaaoi0")

db.PG_CONN_STR = f"postgresql://{PG_USER}:{PG_PASSWORD}@{PG_HOST}:{PG_PORT}/{PG_DB}"

DEFAULT_SYMBOLS = [
    'BNB', 'ETH', 'xyz:GOLD', 'xyz:SILVER', 'SUI',
    'xyz:BRENTOIL', 'mkts:USTECH', 'xyz:CL', 'xyz:XYZ100',
    'HYPE', 'SOL', 'BTC'
]


def get_all_symbols(conn):
    """Get all 1m table names from the database."""
    cur = conn.cursor()
    cur.execute(
        """
        SELECT table_name
        FROM information_schema.tables
        WHERE table_schema = 'public'
          AND table_type = 'BASE TABLE'
          AND table_name LIKE '%_1m'
        ORDER BY table_name
        """
    )
    tables = cur.fetchall()
    cur.close()
    symbols = []
    for (table_name,) in tables:
        symbol = table_name[:-3] if table_name.endswith('_1m') else table_name
        symbols.append(symbol)
    return symbols


def get_last_candle(conn, symbol):
    """Get the most recent 1m candle for a symbol."""
    table = db.sanitize_table_name(symbol, '1m')
    if not db.table_exists(conn, table):
        return None
    cur = conn.cursor()
    cur.execute(f'SELECT * FROM "{table}" ORDER BY timestamp_ms DESC LIMIT 1')
    row = cur.fetchone()
    cur.close()
    return row


def get_nearest_candle(conn, symbol, target_ts):
    """Get the candle nearest to the target timestamp."""
    table = db.sanitize_table_name(symbol, '1m')
    if not db.table_exists(conn, table):
        return None
    cur = conn.cursor()
    cur.execute(
        f'SELECT * FROM "{table}" ORDER BY ABS(timestamp_ms - %s) LIMIT 1',
        (target_ts,)
    )
    row = cur.fetchone()
    cur.close()
    return row


def format_row(symbol, row):
    """Format a candle row for display."""
    if row is None:
        return f"  {symbol:<15} | No data"
    dt = row[0]
    ts = row[1]
    o = row[2]
    h = row[3]
    l = row[4]
    c = row[5]
    v = row[6]
    return (
        f"  {symbol:<15} | "
        f"{dt} | "
        f"{ts} | "
        f"O={o:.2f} H={h:.2f} L={l:.2f} C={c:.2f} | "
        f"Vol={v:.2f}"
    )


def main():
    parser = argparse.ArgumentParser(
        description="Read last 1m candles from PostgreSQL database on NAS."
    )
    parser.add_argument('--all', action='store_true',
                        help='List last candles for all symbols')
    parser.add_argument('--timestamp', type=str, default=None,
                        help='DD-MM-YY or "DD-MM-YY HH:mm", list nearest candle to this date/time')
    args, unknown = parser.parse_known_args()

    symbols = []
    for arg in unknown:
        if arg.startswith('--'):
            symbol = arg[2:]
            symbols.append(symbol)

    target_ts = None
    if args.timestamp:
        try:
            if ' ' in args.timestamp:
                dt = datetime.strptime(args.timestamp, '%d-%m-%y %H:%M')
            else:
                dt = datetime.strptime(args.timestamp, '%d-%m-%y')
            target_ts = int(dt.timestamp() * 1000)
        except ValueError:
            print('Error: Invalid timestamp format. Use DD-MM-YY or "DD-MM-YY HH:mm" (e.g., 05-08-26 or "05-08-26 13:30")')
            sys.exit(1)

    conn = db.get_connection()

    if args.all:
        symbols = get_all_symbols(conn)
        mode = "ALL symbols"
    elif symbols:
        mode = f"specified symbols: {', '.join(symbols)}"
    else:
        symbols = DEFAULT_SYMBOLS
        mode = "default 12 symbols"

    ts_str = f" (nearest to {args.timestamp})" if target_ts else ""
    print(f"\n{'=' * 70}")
    print(f"Last 1m Candles - {mode}{ts_str}")
    print(f"{'=' * 70}")

    for symbol in symbols:
        if target_ts:
            row = get_nearest_candle(conn, symbol, target_ts)
        else:
            row = get_last_candle(conn, symbol)
        print(format_row(symbol, row))

    print(f"{'=' * 70}\n")
    conn.close()


if __name__ == "__main__":
    main()