c7ee5135ae
- FastAPI backend with PostgreSQL database connection - Frontend dashboard with lightweight-charts - Technical indicators (SMA, EMA, RSI, MACD, Bollinger Bands, etc.) - Trading strategy simulation and backtesting - Database connection to NAS at 20.20.20.20:5433 - Development server setup and documentation
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
import asyncio
|
|
import os
|
|
from dotenv import load_dotenv
|
|
import asyncpg
|
|
|
|
load_dotenv()
|
|
|
|
async def test_db_connection():
|
|
"""Test database connection"""
|
|
try:
|
|
conn = await asyncpg.connect(
|
|
host=os.getenv('DB_HOST'),
|
|
port=int(os.getenv('DB_PORT', 5432)),
|
|
database=os.getenv('DB_NAME'),
|
|
user=os.getenv('DB_USER'),
|
|
password=os.getenv('DB_PASSWORD'),
|
|
)
|
|
|
|
version = await conn.fetchval('SELECT version()')
|
|
print(f"[OK] Database connected successfully!")
|
|
print(f" Host: {os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}")
|
|
print(f" Database: {os.getenv('DB_NAME')}")
|
|
print(f" User: {os.getenv('DB_USER')}")
|
|
print(f" PostgreSQL: {version[:50]}...")
|
|
|
|
# Check if tables exist
|
|
tables = await conn.fetch("""
|
|
SELECT table_name FROM information_schema.tables
|
|
WHERE table_schema = 'public'
|
|
ORDER BY table_name
|
|
""")
|
|
|
|
table_names = [row['table_name'] for row in tables]
|
|
print(f"\n[OK] Found {len(table_names)} tables:")
|
|
for table in table_names:
|
|
print(f" - {table}")
|
|
|
|
# Check candles count
|
|
if 'candles' in table_names:
|
|
count = await conn.fetchval('SELECT COUNT(*) FROM candles')
|
|
latest_time = await conn.fetchval("""
|
|
SELECT MAX(time) FROM candles
|
|
WHERE time > NOW() - INTERVAL '7 days'
|
|
""")
|
|
print(f"\n[OK] Candles table has {count} total records")
|
|
if latest_time:
|
|
print(f" Latest candle (last 7 days): {latest_time}")
|
|
|
|
await conn.close()
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"[FAIL] Database connection failed:")
|
|
print(f" Error: {e}")
|
|
print(f"\nCheck:")
|
|
print(f" 1. NAS is reachable at {os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}")
|
|
print(f" 2. PostgreSQL is running")
|
|
print(f" 3. Database '{os.getenv('DB_NAME')}' exists")
|
|
print(f" 4. User '{os.getenv('DB_USER')}' has access")
|
|
return False
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(test_db_connection()) |