hyper/list_latest_candles.py

import argparse
import json
import logging
import os
import sys
from datetime import datetime, timezone
from contextlib import closing

import psycopg2

from logging_utils import setup_logging

DEFAULT_DB_PATH = os.environ.get(
    "PG_CONN_STR",
    "postgresql://hyper:hyper@localhost:5432/hyper"
)


def load_coins():
    """Load the list of all coins from the local coin_precision.json file."""
    coin_file = "_data/coin_precision.json"
    try:
        with open(coin_file, 'r') as f:
            return list(json.load(f).keys())
    except FileNotFoundError:
        logging.error(f"'{coin_file}' not found. Please run list_coins.py first.")
        sys.exit(1)
    except (IOError, json.JSONDecodeError) as e:
        logging.error(f"Failed to load or parse '{coin_file}': {e}")
        sys.exit(1)


def get_latest_candle(conn, coin, interval="1m"):
    """
    Query the database for the most recent candle for a given coin.

    Returns a dict with keys: datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades
    or None if the table doesn't exist or has no rows.
    """
    table_name = f"{coin.replace(':', '_')}_{interval}"
    try:
        with closing(conn.cursor()) as cur:
            cur.execute(f'SELECT 1 FROM information_schema.tables WHERE table_name = %s', (table_name,))
            if not cur.fetchone()[0]:
                return None

            cur.execute(
                f'SELECT datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades '
                f'FROM "{table_name}" ORDER BY timestamp_ms DESC LIMIT 1'
            )
            row = cur.fetchone()
            if row is None:
                return None

            return {
                "datetime_utc": row[0],
                "timestamp_ms": row[1],
                "open": row[2],
                "high": row[3],
                "low": row[4],
                "close": row[5],
                "volume": row[6],
                "number_of_trades": row[7],
            }
    except Exception as e:
        logging.debug(f"Could not get latest candle for {coin} ({interval}): {e}")
        return None


def list_latest_candles(coins, interval="1m", db_path=None):
    """
    Fetch and display the newest candle for every coin in the list.
    """
    if db_path is None:
        db_path = DEFAULT_DB_PATH

    conn = psycopg2.connect(db_path)

    results = []
    for coin in coins:
        candle = get_latest_candle(conn, coin, interval)
        if candle is not None:
            results.append((coin, candle))
        else:
            results.append((coin, None))

    conn.close()

    print(f"\n--- Newest {interval} Candles for All Symbols ---")
    print(f"Total symbols: {len(coins)} | Symbols with data: {sum(1 for _, c in results if c is not None)}")
    print(f"Generated at: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}")
    print("-" * 120)
    print(f"{'Coin':<16} | {'Datetime (UTC)':<22} | {'Open':>12} | {'High':>12} | {'Low':>12} | {'Close':>12} | {'Volume':>12}")
    print("-" * 120)

    for coin, candle in results:
        if candle is not None:
            dt = candle["datetime_utc"].strftime('%Y-%m-%d %H:%M:%S') if candle["datetime_utc"] else "N/A"
            o = f"{candle['open']:.4f}" if candle['open'] is not None else "N/A"
            h = f"{candle['high']:.4f}" if candle['high'] is not None else "N/A"
            l = f"{candle['low']:.4f}" if candle['low'] is not None else "N/A"
            c = f"{candle['close']:.4f}" if candle['close'] is not None else "N/A"
            v = f"{candle['volume']:.4f}" if candle['volume'] is not None else "N/A"
            print(f"{coin:<16} | {dt:<22} | {o:>12} | {h:>12} | {l:>12} | {c:>12} | {v:>12}")
        else:
            print(f"{coin:<16} | {'(no data)':<22} | {'':>12} | {'':>12} | {'':>12} | {'':>12} | {'':>12}")

    print("-" * 120)
    print(f"Symbols without data: {sum(1 for _, c in results if c is None)}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="List the newest 1-minute candle for all symbols from the database."
    )
    parser.add_argument(
        "--interval",
        default="1m",
        help="Candle interval to query (default: 1m)."
    )
    parser.add_argument(
        "--db",
        default=None,
        help="PostgreSQL connection string (default: from PG_CONN_STR env or localhost)."
    )
    parser.add_argument(
        "--log-level",
        default="off",
        choices=['off', 'normal', 'debug'],
        help="Set the logging level."
    )
    args = parser.parse_args()

    setup_logging(args.log_level, 'ListLatestCandles')

    coins = load_coins()
    list_latest_candles(coins, interval=args.interval, db_path=args.db)