#!/usr/bin/env python3
"""
Data fetchers for the dashboard.
Collects data from bot, cron, database, logs, and system.
"""
import os
import subprocess
import sqlite3
import glob
import json
from datetime import datetime
from typing import Optional, Dict, Any
def get_bot_status() -> Dict[str, Any]:
"""Check if trading bot is running and get uptime."""
try:
# Check for trading_bot process
result = subprocess.run(
["pgrep", "-f", "trading_bot"],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0:
pids = result.stdout.strip().split('\n')
# Get uptime of first process
first_pid = pids[0]
uptime_result = subprocess.run(
["ps", "-p", first_pid, "-o", "lstart"],
capture_output=True, text=True, timeout=5
)
uptime = uptime_result.stdout.strip().split('\n')[-1].strip() if uptime_result.stdout else "Unknown"
return {
"running": True,
"pid": first_pid,
"uptime": uptime,
"mode": "Paper Trading" # Could detect from config
}
else:
return {
"running": False,
"pid": None,
"uptime": "Not running",
"mode": "Stopped"
}
except Exception as e:
return {
"running": False,
"pid": None,
"uptime": f"Error: {str(e)}",
"mode": "Unknown"
}
def get_cron_status() -> Dict[str, Any]:
"""List active cron jobs related to the bot."""
try:
# Get cron jobs from Hermes cron system
cron_file = os.path.expanduser("~/.hermes/cron/jobs.json")
if os.path.exists(cron_file):
with open(cron_file, 'r') as f:
jobs_data = json.load(f)
# Filter for relevant jobs
bot_jobs = []
for job in jobs_data.get('jobs', []):
if job.get('enabled') and not job.get('paused_at'):
bot_jobs.append({
"name": job.get('name', 'Unknown'),
"schedule": job.get('schedule', 'Unknown'),
"next_run": job.get('next_run_at', 'Unknown'),
"last_run": job.get('last_run_at', 'Never'),
"status": "active"
})
return {
"jobs": bot_jobs,
"count": len(bot_jobs)
}
else:
return {
"jobs": [],
"count": 0,
"error": "Cron file not found"
}
except Exception as e:
return {
"jobs": [],
"count": 0,
"error": str(e)
}
def get_trade_summary(days: int = 7) -> Dict[str, Any]:
"""Get trading summary from signals database."""
db_path = os.path.expanduser("~/bitcoin-trading-bot/trading_bot.db")
if not os.path.exists(db_path):
return {
"error": "Database not found",
"total_signals": 0,
"total_trades": 0,
"winning_trades": 0,
"losing_trades": 0,
"win_rate": 0,
"total_pnl": 0,
"latest_signal": None
}
try:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Get signal count
cursor.execute("SELECT COUNT(*) as count FROM signals")
total_signals = cursor.fetchone()['count']
# Get trade summary
cursor.execute("""
SELECT
COUNT(*) as total,
SUM(CASE WHEN status = 'closed' THEN 1 ELSE 0 END) as closed,
SUM(CASE WHEN pnl > 0 THEN 1 ELSE 0 END) as wins,
SUM(CASE WHEN pnl < 0 THEN 1 ELSE 0 END) as losses,
COALESCE(SUM(pnl), 0) as total_pnl
FROM trades
""")
trade_row = cursor.fetchone()
total_trades = trade_row['total'] or 0
closed_trades = trade_row['closed'] or 0
winning_trades = trade_row['wins'] or 0
losing_trades = trade_row['losses'] or 0
total_pnl = trade_row['total_pnl'] or 0
win_rate = (winning_trades / closed_trades * 100) if closed_trades > 0 else 0
# Get latest signal
cursor.execute("""
SELECT symbol, timestamp, state, tau, eps, signal_type
FROM signals
ORDER BY created_at DESC LIMIT 1
""")
latest_signal = cursor.fetchone()
conn.close()
return {
"total_signals": total_signals,
"total_trades": total_trades,
"closed_trades": closed_trades,
"winning_trades": winning_trades,
"losing_trades": losing_trades,
"win_rate": round(win_rate, 2),
"total_pnl": round(total_pnl, 2),
"latest_signal": dict(latest_signal) if latest_signal else None
}
except Exception as e:
return {
"error": str(e),
"total_signals": 0,
"total_trades": 0,
"winning_trades": 0,
"losing_trades": 0,
"win_rate": 0,
"total_pnl": 0,
"latest_signal": None
}
def get_system_health() -> Dict[str, Any]:
"""Get system resource usage."""
health = {}
# Disk usage - check both SSD and HDD
try:
mounts = [('/', 'SSD'), ('/mnt/storage', 'HDD')]
disks = []
for mount, label in mounts:
result = subprocess.run(["df", "-h", mount], capture_output=True, text=True, timeout=5)
lines = result.stdout.strip().split('\n')
if len(lines) >= 2:
parts = lines[1].split()
disks.append({
'label': label,
'mount': mount,
'used': parts[2],
'total': parts[1],
'percent': parts[4]
})
health['disks'] = disks
except Exception as e:
health['disk_error'] = str(e)
# Memory usage
try:
with open('/proc/meminfo', 'r') as f:
meminfo = {}
for line in f:
parts = line.split()
if len(parts) >= 2:
key = parts[0].rstrip(':')
value = int(parts[1]) # in kB
meminfo[key] = value
total = meminfo.get('MemTotal', 0) / 1024 # MB
available = meminfo.get('MemAvailable', 0) / 1024 # MB
used = total - available
percent = (used / total * 100) if total > 0 else 0
health['memory'] = {
'total': f"{total:.0f} MB",
'used': f"{used:.0f} MB",
'available': f"{available:.0f} MB",
'percent': f"{percent:.1f}%"
}
except Exception as e:
health['memory_error'] = str(e)
# CPU usage (more efficient method)
try:
# Read /proc/stat directly instead of running top
with open('/proc/stat', 'r') as f:
line = f.readline()
parts = line.split()
# cpu user nice system idle iowait irq softirq steal guest guest_nice
if len(parts) >= 5:
user = int(parts[1])
nice = int(parts[2])
system = int(parts[3])
idle = int(parts[4])
total = user + nice + system + idle
cpu_usage = ((total - idle) / total * 100) if total > 0 else 0
health['cpu'] = {
'idle': f"{100 - cpu_usage:.1f}%",
'usage': f"{cpu_usage:.1f}%"
}
except Exception as e:
health['cpu_error'] = str(e)
# GPU Power (NVIDIA)
try:
result = subprocess.run(
["nvidia-smi", "--query-gpu=power.draw,power.limit", "--format=csv,noheader"],
capture_output=True, text=True, timeout=10
)
if result.returncode == 0:
parts = result.stdout.strip().split(',')
power_draw = float(parts[0].replace(' W', '').strip())
power_limit = float(parts[1].replace(' W', '').strip())
health['gpu_power'] = {
'draw': f"{power_draw:.1f} W",
'limit': f"{power_limit:.1f} W",
'percent': f"{(power_draw / power_limit * 100):.1f}%"
}
# Calculate estimated total system power
# Get CPU frequency for better power estimation
try:
with open('/sys/devices/system/cpu/cpufreq/policy0/scaling_cur_freq', 'r') as f:
cur_freq = int(f.read().strip()) / 1000 # Convert to MHz
max_freq = 5600 # Ryzen 9 9900X max boost (MHz)
freq_ratio = cur_freq / max_freq
# AMD Ryzen 9 9900X power model:
# Base: ~65W TDP at base frequency (~4.2GHz)
# Peak: ~170W PPT at boost (~5.6GHz)
# Plus frequency scaling overhead
base_cpu_power = 65
boost_cpu_power = 170
cpu_power_estimate = base_cpu_power + (boost_cpu_power - base_cpu_power) * freq_ratio
# Add CPU usage factor (workload intensity)
cpu_power_estimate = cpu_power_estimate * (0.5 + 0.5 * (cpu_usage / 100))
except:
# Fallback to usage-based estimate
base_power = 50 # Base system power (motherboard, RAM, drives, fans, etc.)
cpu_estimate = cpu_usage * 1.7 # ~170W at 100% CPU for Ryzen 9 9900X
total_estimated = base_power + cpu_estimate + power_draw
health['total_power'] = {
'estimated': f"{total_estimated:.1f} W",
'gpu': f"{power_draw:.1f} W",
'cpu_estimated': f"{cpu_estimate:.1f} W",
'base': f"{base_power} W"
}
cpu_power_estimate = cpu_estimate
cur_freq = None
# Base system power (motherboard, RAM, drives, fans, etc.)
base_power = 50
total_estimated = base_power + cpu_power_estimate + power_draw
health['total_power'] = {
'estimated': f"{total_estimated:.1f} W",
'gpu': f"{power_draw:.1f} W",
'cpu_estimated': f"{cpu_power_estimate:.1f} W",
'base': f"{base_power} W"
}
if cur_freq:
health['cpu_freq'] = {
'current': f"{cur_freq:.0f} MHz",
'max': f"{max_freq} MHz",
'ratio': f"{(cur_freq / max_freq * 100):.1f}%"
}
except Exception as e:
health['gpu_error'] = str(e)
return health
def get_account_value() -> Dict[str, Any]:
"""Get paper trading account value from settings and positions.
Returns initial portfolio value ($10k for paper trading) plus unrealized PnL.
"""
# Paper trading initial value
initial_portfolio = 10000
# Get current positions and calculate unrealized PnL
db_path = os.path.expanduser("~/bitcoin-trading-bot/trading_bot.db")
unrealized_pnl = 0
positions = []
if os.path.exists(db_path):
try:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Get open positions with metadata
cursor.execute("""
SELECT symbol, entry_price, metadata
FROM signals
WHERE signal_type = 'entry' AND executed = 1
AND (execution_note LIKE '%open%' OR execution_note LIKE '%Holding%')
""")
for row in cursor.fetchall():
metadata = row['metadata']
if metadata:
import json
pos_data = json.loads(metadata)
positions.append({
'symbol': row['symbol'],
'entry_price': row['entry_price'],
'position_value': pos_data.get('position_value', 0),
'position_size': pos_data.get('position_size', 0),
})
conn.close()
except Exception as e:
print(f"Error fetching positions for account value: {e}")
return {
"initial_portfolio": initial_portfolio,
"current_value": initial_portfolio, # Paper trading: fixed value
"unrealized_pnl": unrealized_pnl,
"positions": positions,
"mode": "Paper Trading"
}
def get_positions_held() -> Dict[str, Any]:
"""Get currently held positions from signals database.
Shows all executed entry signals. Uses kelly fraction to estimate
position value when entry_price is not available.
"""
db_path = os.path.expanduser("~/bitcoin-trading-bot/trading_bot.db")
if not os.path.exists(db_path):
return {
"error": "Database not found",
"positions": [],
"total_value": 0,
"count": 0
}
initial_portfolio = 10000 # Paper trading value
try:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Check if metadata column exists
cursor.execute("PRAGMA table_info(signals)")
columns = [row[1] for row in cursor.fetchall()]
has_metadata = 'metadata' in columns
# Get ALL executed entry signals (matches bot's status.py behavior)
if has_metadata:
cursor.execute("""
SELECT symbol, entry_price, stop_loss, take_profit,
kelly_fraction, execution_note, created_at, metadata
FROM signals
WHERE signal_type = 'entry' AND executed = 1
ORDER BY created_at DESC
""")
else:
cursor.execute("""
SELECT symbol, entry_price, stop_loss, take_profit,
kelly_fraction, execution_note, created_at
FROM signals
WHERE signal_type = 'entry' AND executed = 1
ORDER BY created_at DESC
""")
positions = []
for row in cursor.fetchall():
pos = dict(row)
# Get actual position value from metadata or estimate from kelly fraction
position_value = 0
# Try to get from metadata first (if column exists)
if has_metadata and pos.get('metadata'):
try:
import json
meta = json.loads(pos['metadata']) if isinstance(pos['metadata'], str) else pos['metadata']
position_value = meta.get('position_value', 0)
except (json.JSONDecodeError, TypeError):
pass
# Fallback: estimate based on kelly fraction of portfolio
# (This matches the bot's status.py behavior)
if position_value == 0 and pos.get('kelly_fraction'):
kelly = pos.get('kelly_fraction') or 0
position_value = initial_portfolio * kelly
positions.append({
"symbol": pos.get('symbol', 'Unknown'),
"entry_price": pos.get('entry_price'),
"stop_loss": pos.get('stop_loss'),
"take_profit": pos.get('take_profit'),
"kelly_fraction": pos.get('kelly_fraction') or 0,
"estimated_value": position_value,
"created_at": pos.get('created_at')
})
conn.close()
# Total value is the portfolio value (paper trading)
return {
"positions": positions,
"total_value": initial_portfolio,
"count": len(positions),
"initial_portfolio": initial_portfolio
}
except Exception as e:
return {
"error": str(e),
"positions": [],
"total_value": initial_portfolio,
"count": 0
}
def get_recent_logs(count: int = 5) -> Dict[str, Any]:
"""Get recent log entries from trading bot."""
try:
# Find latest log file
log_pattern = "/home/vincent/bitcoin-trading-bot/logs/trading_bot_*.log"
log_files = glob.glob(log_pattern)
if not log_files:
return {
"logs": [],
"error": "No log files found"
}
# Get latest log file
latest_log = max(log_files)
# Read last N lines
with open(latest_log, 'r') as f:
lines = f.readlines()
recent = [line.strip() for line in lines[-count:] if line.strip()]
return {
"logs": recent,
"file": os.path.basename(latest_log)
}
except Exception as e:
return {
"logs": [],
"error": str(e)
}
def get_all_data() -> Dict[str, Any]:
"""Get all dashboard data in one call."""
return {
"timestamp": datetime.now().isoformat(),
"account": get_account_value(),
"bot": get_bot_status(),
"cron": get_cron_status(),
"trades": get_trade_summary(),
"positions": get_positions_held(),
"system": get_system_health(),
"logs": get_recent_logs()
}