p1otek/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add testing scripts for DB connection and candle reading

Browse Source 
- test_db_connection.py: 5-stage DB connection test (TCP, auth, schema, data, db.py integration)
- read_candles.py: Read 1m candles with --all, --[symbol], --timestamp support
- README.md: Usage instructions for testing scripts
- .env.example: Template for connection configuration

Default symbols updated to match LiveCandleFetcher subscriptions (BNB, ETH, xyz:GOLD, etc.)
Colon-containing symbols handled via db.sanitize_table_name()
This commit is contained in:
DiTus
2026-08-05 20:33:54 +02:00
parent 1a95fe1caa
commit 7552cdfd57
4 changed files with 627 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
testing/.env.example Normal file
View File

@ -0,0 +1,9 @@
# Database connection configuration for testing
# Copy this file to .env and adjust as needed.
PG_HOST=20.20.20.20
PG_PORT=5433
PG_DB=hyper
PG_USER=hyper
PG_PASSWORD=your_password_here
PG_TIMEOUT=10

143
testing/README.md Normal file
View File

@ -0,0 +1,143 @@
# Candle Reader - Testing Instructions
## Overview
`read_candles.py` is a test script that reads 1m candle data from the PostgreSQL database running in Docker on the Synology NAS (`20.20.20.20`).
## Prerequisites
1. **Python 3.11+** with `psycopg2-binary` installed:
```bash
pip install psycopg2-binary
```
2. **Network access** to the NAS at `20.20.20.20` on port `5433`.
3. **Docker containers running** on the NAS:
```bash
docker-compose up -d
```
## Configuration
Connection parameters are loaded from `testing/.env` (git-ignored). Copy from the example if needed:
```bash
cp testing/.env.example testing/.env
```
Edit `testing/.env` to match your NAS setup:
| Variable | Default | Description |
|----------|---------|-------------|
| `PG_HOST` | `20.20.20.20` | NAS IP address |
| `PG_PORT` | `5433` | PostgreSQL port (host) |
| `PG_DB` | `hyper` | Database name |
| `PG_USER` | `hyper` | Database user |
| `PG_PASSWORD` | `kaqpaaoi0` | Database password |
| `PG_TIMEOUT` | `10` | Connection timeout (seconds) |
## Usage
```bash
python testing/read_candles.py [OPTIONS] [--SYMBOL ...]
```
### Options
| Flag | Description |
|------|-------------|
| `--all` | List last candles for **all** symbols (411+ tables) |
| `--timestamp DD-MM-YY` | List nearest candle to the given date (start of day) |
| `--timestamp "DD-MM-YY HH:mm"` | List nearest candle to the given date and time |
| `-h, --help` | Show help message |
### Dynamic Symbol Flags
Any `--SYMBOL` flag filters to that specific symbol(s). Multiple symbols can be combined.
| Flag | Description |
|------|-------------|
| `--BTC` | Only BTC candles |
| `--BTC --ETH` | BTC and ETH candles only |
| `--xyz:GOLD` | Only xyz:GOLD candles (colon symbols supported) |
| `--xyz:BRENTOIL` | Only xyz:BRENTOIL candles |
### Default Behavior
With no flags, the script reads the last 1m candle for 12 default symbols:
`BNB`, `ETH`, `xyz:GOLD`, `xyz:SILVER`, `SUI`, `xyz:BRENTOIL`, `mkts:USTECH`, `xyz:CL`, `xyz:XYZ100`, `HYPE`, `SOL`, `BTC`
Symbols containing colons (e.g., `xyz:GOLD`) are automatically sanitized to PostgreSQL-safe table names (e.g., `xyz_GOLD_1m`).
## Examples
### 1. Last candles for 12 default symbols
```bash
python testing/read_candles.py
```
### 2. Last candles for all symbols
```bash
python testing/read_candles.py --all
```
### 3. Last candle for a single symbol
```bash
python testing/read_candles.py --BTC
```
### 4. Last candles for multiple symbols
```bash
python testing/read_candles.py --BTC --ETH --SOL
```
### 5. Last candles for colon-containing symbols
```bash
python testing/read_candles.py --xyz:GOLD --xyz:BRENTOIL
```
### 5. Nearest candle to a date (start of day)
```bash
python testing/read_candles.py --BTC --timestamp 05-08-26
```
### 6. Nearest candle to a date and time
```bash
python testing/read_candles.py --BTC --timestamp "05-08-26 13:30"
```
### 7. Combine --all with --timestamp
```bash
python testing/read_candles.py --all --timestamp "05-08-26 13:15"
```
## Output Format
```
======================================================================
Last 1m Candles - default 12 symbols
======================================================================
BTC | 2026-08-05 13:15:00 | 1785935700000 | O=64116.00 H=64116.00 L=64096.00 C=64104.00 | Vol=2.19
ETH | 2026-08-05 13:15:00 | 1785935700000 | O=1868.00 H=1868.00 L=1867.50 C=1867.60 | Vol=32.26
...
======================================================================
```
Columns: `Symbol | datetime_utc | timestamp_ms | O(open) H(high) L(low) C(close) | Vol(volume)`
## Troubleshooting
### Connection refused
- Ensure Docker containers are running: `docker-compose up -d`
- Check NAS firewall allows port 5433
- Verify `PG_HOST` in `testing/.env`
### No data for a symbol
- The symbol may not have a table (e.g., `MATIC_1m` may not exist)
- Use `--all` to see which tables are available
- Ensure the symbol name matches exactly (case-sensitive, colons included)
### Invalid timestamp format
- Use `DD-MM-YY` (e.g., `05-08-26`)
- Or `"DD-MM-YY HH:mm"` (e.g., `"05-08-26 13:30"`)

179
testing/read_candles.py Normal file
View File

@ -0,0 +1,179 @@
#!/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()

296
testing/test_db_connection.py Normal file
View File

@ -0,0 +1,296 @@
#!/usr/bin/env python3
"""
Database Connection Test Script
Tests connectivity from the local laptop to the PostgreSQL database
running in Docker on the Synology NAS at 20.20.20.20.
Tests performed:
1. TCP connectivity to the NAS port
2. PostgreSQL authentication
3. Schema inspection (list tables)
4. Data verification (row counts, latest records)
5. Integration with the project's db.py module
Usage:
python testing/test_db_connection.py
python testing/test_db_connection.py --host 20.20.20.20 --port 5433
python testing/test_db_connection.py --verbose
Environment variables (or .env file in this directory):
PG_HOST - NAS IP address (default: 20.20.20.20)
PG_PORT - PostgreSQL port on NAS (default: 5433)
PG_DB - Database name (default: hyper)
PG_USER - Database user (default: hyper)
PG_PASSWORD - Database password (default: kaqpaaoi0)
PG_TIMEOUT - Connection timeout in seconds (default: 10)
"""
import os
import sys
import socket
import argparse
import logging
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)
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")
PG_TIMEOUT = int(os.environ.get("PG_TIMEOUT", "10"))
class TestResult:
"""Track pass/fail results across all test stages."""
def __init__(self):
self.passed = 0
self.failed = 0
self.results = []
def add(self, name, success, message):
self.results.append((name, success, message))
if success:
self.passed += 1
else:
self.failed += 1
def summary(self):
total = self.passed + self.failed
return (
f"\n{'=' * 60}\n"
f"Results: {self.passed}/{total} passed, {self.failed} failed\n"
f"{'=' * 60}"
)
def test_tcp_connectivity(host, port, timeout):
"""Stage 1: Verify TCP reachability to the NAS port."""
try:
sock = socket.create_connection((host, port), timeout=timeout)
sock.close()
return True, f"TCP connection to {host}:{port} succeeded"
except socket.timeout:
return False, f"TCP connection to {host}:{port} timed out after {timeout}s"
except ConnectionRefusedError:
return False, (
f"Connection refused to {host}:{port}. "
"Port may be closed or Synology DSM firewall is blocking it."
)
except socket.gaierror:
return False, f"Could not resolve hostname: {host}"
except Exception as e:
return False, f"TCP connection error: {e}"
def test_pg_auth(conn_str, timeout):
"""Stage 2: Verify PostgreSQL authentication with credentials."""
try:
import psycopg2
except ImportError:
return False, (
"psycopg2 is not installed. "
"Install it with: pip install psycopg2-binary"
)
try:
conn = psycopg2.connect(conn_str, connect_timeout=timeout)
cur = conn.cursor()
cur.execute("SELECT 1")
cur.fetchone()
cur.close()
conn.close()
masked = conn_str.split("@")[1] if "@" in conn_str else conn_str
return True, f"PostgreSQL authentication successful ({masked})"
except psycopg2.OperationalError as e:
return False, f"PostgreSQL connection failed: {e}"
except Exception as e:
return False, f"PostgreSQL error: {e}"
def test_schema_inspection(conn_str):
"""Stage 3: List tables in the public schema."""
try:
import psycopg2
conn = psycopg2.connect(conn_str)
cur = conn.cursor()
cur.execute(
"""
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_type = 'BASE TABLE'
ORDER BY table_name
"""
)
tables = cur.fetchall()
cur.close()
conn.close()
table_names = [t[0] for t in tables]
sample = table_names[:5] if table_names else "none"
return True, f"Found {len(table_names)} tables. Sample: {sample}"
except Exception as e:
return False, f"Schema inspection failed: {e}"
def test_data_verification(conn_str):
"""Stage 4: Check row count and latest record for a table."""
try:
import psycopg2
conn = psycopg2.connect(conn_str)
cur = conn.cursor()
cur.execute(
"""
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_type = 'BASE TABLE'
ORDER BY table_name
LIMIT 1
"""
)
result = cur.fetchone()
if not result:
cur.close()
conn.close()
return False, "No tables found in database"
table_name = result[0]
cur.execute(f'SELECT COUNT(*) FROM "{table_name}"')
count = cur.fetchone()[0]
cur.execute(
f'SELECT timestamp_ms FROM "{table_name}" '
f'ORDER BY timestamp_ms DESC LIMIT 1'
)
latest = cur.fetchone()
cur.close()
conn.close()
if count == 0:
return True, f"Table '{table_name}' exists but is empty (0 rows)"
latest_ts = latest[0] if latest else "N/A"
return True, f"Table '{table_name}': {count} rows, latest timestamp_ms={latest_ts}"
except Exception as e:
return False, f"Data verification failed: {e}"
def test_db_py_integration(conn_str):
"""Stage 5: Verify the project's db.py module works with the remote connection."""
try:
import db
except ImportError as e:
return False, (
f"Could not import db.py: {e}. "
"Ensure the project root is accessible."
)
try:
old_conn_str = db.PG_CONN_STR
db.PG_CONN_STR = conn_str
conn = db.get_connection()
cur = conn.cursor()
cur.execute("SELECT 1")
cur.fetchone()
cur.close()
conn.close()
db.PG_CONN_STR = old_conn_str
return True, "db.py get_connection() works with remote connection string"
except Exception as e:
return False, f"db.py integration failed: {e}"
def main():
parser = argparse.ArgumentParser(
description="Test PostgreSQL database connection from laptop to NAS Docker container."
)
parser.add_argument("--host", default=PG_HOST, help=f"NAS IP address (default: {PG_HOST})")
parser.add_argument("--port", type=int, default=PG_PORT, help=f"PostgreSQL port (default: {PG_PORT})")
parser.add_argument("--db", default=PG_DB, help=f"Database name (default: {PG_DB})")
parser.add_argument("--user", default=PG_USER, help=f"Database user (default: {PG_USER})")
parser.add_argument("--password", default=PG_PASSWORD, help=f"Database password (default: {PG_PASSWORD})")
parser.add_argument("--timeout", type=int, default=PG_TIMEOUT, help=f"Connection timeout in seconds (default: {PG_TIMEOUT})")
parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
parser.add_argument("--skip-db-py", action="store_true", help="Skip db.py integration test")
args = parser.parse_args()
host = args.host
port = args.port
db_name = args.db
user = args.user
password = args.password
timeout = args.timeout
conn_str = f"postgresql://{user}:{password}@{host}:{port}/{db_name}"
log_level = logging.DEBUG if args.verbose else logging.INFO
logging.basicConfig(level=log_level, format="%(asctime)s - %(levelname)s - %(message)s")
print(f"\n{'=' * 60}")
print("Database Connection Test")
print(f"{'=' * 60}")
print(f"Target: {host}:{port}")
print(f"Database: {db_name}")
print(f"User: {user}")
print(f"Timeout: {timeout}s")
print(f"Connection: postgresql://{user}:***@{host}:{port}/{db_name}")
print(f"{'=' * 60}\n")
results = TestResult()
print("[1/5] Testing TCP connectivity...")
success, message = test_tcp_connectivity(host, port, timeout)
print(f" [{'PASS' if success else 'FAIL'}] {message}")
results.add("TCP connectivity", success, message)
print("\n[2/5] Testing PostgreSQL authentication...")
success, message = test_pg_auth(conn_str, timeout)
print(f" [{'PASS' if success else 'FAIL'}] {message}")
results.add("PostgreSQL authentication", success, message)
if not success:
print("\n Authentication failed. Cannot proceed with further tests.")
print(" Possible causes:")
print(" - Wrong credentials")
print(" - PostgreSQL not running or port not mapped (docker-compose up -d)")
print(" - Synology DSM firewall blocking port 5433")
print(f"\n{results.summary()}")
sys.exit(1)
print("\n[3/5] Testing schema inspection...")
success, message = test_schema_inspection(conn_str)
print(f" [{'PASS' if success else 'FAIL'}] {message}")
results.add("Schema inspection", success, message)
print("\n[4/5] Testing data verification...")
success, message = test_data_verification(conn_str)
print(f" [{'PASS' if success else 'FAIL'}] {message}")
results.add("Data verification", success, message)
if not args.skip_db_py:
print("\n[5/5] Testing db.py integration...")
success, message = test_db_py_integration(conn_str)
print(f" [{'PASS' if success else 'FAIL'}] {message}")
results.add("db.py integration", success, message)
print(f"\n{'=' * 60}")
print("Summary")
print(f"{'=' * 60}")
for name, success, _ in results.results:
print(f" [{'PASS' if success else 'FAIL'}] {name}")
print(f"\nTotal: {results.passed}/{results.passed + results.failed} passed")
print(f"{'=' * 60}\n")
sys.exit(0 if results.failed == 0 else 1)
if __name__ == "__main__":
main()