#!/usr/bin/env python3
"""
Dashboard Flask Application
Provides web interface for monitoring trading bot and system status.
"""
import os
import sys
import time
from functools import wraps
from flask import Flask, render_template, jsonify, request
# Global cleanup state
_last_cleanup_time = time.time()
_cleanup_interval = 60 # Clean up every 60 seconds
# Add data directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'data'))
from fetcher import get_all_data, get_bot_status, get_trade_summary
app = Flask(__name__)
# Simple basic auth credentials (change in production)
AUTH_USERNAME = 'admin'
AUTH_PASSWORD = 'admin'
def require_auth(f):
"""Decorator to require basic authentication."""
@wraps(f)
def decorated(*args, **kwargs):
auth = request.authorization
if not auth or not (auth.username == AUTH_USERNAME and auth.password == AUTH_PASSWORD):
return jsonify({'error': 'Authentication required'}), 401, {
'WWW-Authenticate': 'Basic realm="Login Required"'
}
return f(*args, **kwargs)
return decorated
def rate_limit(max_requests: int = 10, window_seconds: int = 60):
"""Simple in-memory rate limiter decorator with periodic cleanup.
Args:
max_requests: Maximum requests allowed in the window
window_seconds: Time window in seconds
"""
# In-memory storage: {ip: [timestamp, ...]}
rate_limit_store = {}
def _cleanup_stale_ips():
"""Remove IPs with no activity in the last window_seconds.
Only runs if _cleanup_interval seconds have passed since last cleanup.
This is lazy cleanup - triggered by requests, not a background thread.
"""
global _last_cleanup_time
current_time = time.time()
# Skip cleanup if not enough time has passed
if current_time - _last_cleanup_time < _cleanup_interval:
return
_last_cleanup_time = current_time
window_start = current_time - window_seconds
# Find and remove IPs with no recent activity
stale_ips = [
ip for ip, timestamps in rate_limit_store.items()
if not timestamps or (timestamps and timestamps[-1] < window_start)
]
for ip in stale_ips:
del rate_limit_store[ip]
if stale_ips:
print(f"🧹 Cleaned up {len(stale_ips)} stale IP(s) from rate limiter")
def decorator(f):
@wraps(f)
def decorated(*args, **kwargs):
client_ip = request.remote_addr
# Periodic cleanup of stale IPs (lazy, non-blocking)
_cleanup_stale_ips()
# Clean old entries and get current count
current_time = time.time()
window_start = current_time - window_seconds
if client_ip not in rate_limit_store:
rate_limit_store[client_ip] = []
# Filter to only requests within window
rate_limit_store[client_ip] = [
ts for ts in rate_limit_store[client_ip]
if ts > window_start
]
# Check if rate limited
if len(rate_limit_store[client_ip]) >= max_requests:
return jsonify({
'error': 'Rate limit exceeded',
'retry_after': window_seconds
}), 429, {
'X-RateLimit-Limit': str(max_requests),
'X-RateLimit-Remaining': '0',
'X-RateLimit-Reset': str(int(current_time + window_seconds))
}
# Record this request
rate_limit_store[client_ip].append(current_time)
remaining = max_requests - len(rate_limit_store[client_ip])
# Add rate limit headers to response
result = f(*args, **kwargs)
if isinstance(result, tuple):
response, status, headers = result if len(result) == 3 else (result[0], result[1], {})
headers['X-RateLimit-Limit'] = str(max_requests)
headers['X-RateLimit-Remaining'] = str(remaining)
headers['X-RateLimit-Reset'] = str(int(current_time + window_seconds))
return response, status, headers
return result
return decorated
return decorator
@app.route('/')
@require_auth
def index():
"""Serve main dashboard page."""
return render_template('index.html')
@app.route('/api/status')
@require_auth
@rate_limit(max_requests=10, window_seconds=60)
def api_status():
"""API endpoint for all dashboard data."""
try:
data = get_all_data()
return jsonify({
"success": True,
"data": data
})
except Exception as e:
return jsonify({
"success": False,
"error": str(e)
}), 500
@app.route('/api/bot')
@require_auth
@rate_limit(max_requests=10, window_seconds=60)
def api_bot():
"""API endpoint for bot status only."""
try:
status = get_bot_status()
return jsonify({
"success": True,
"data": status
})
except Exception as e:
return jsonify({
"success": False,
"error": str(e)
}), 500
@app.route('/api/trades')
@require_auth
@rate_limit(max_requests=10, window_seconds=60)
def api_trades():
"""API endpoint for trade summary only."""
try:
summary = get_trade_summary()
return jsonify({
"success": True,
"data": summary
})
except Exception as e:
return jsonify({
"success": False,
"error": str(e)
}), 500
@app.route('/api/health')
def api_health():
"""Health check endpoint (public, no auth)."""
return jsonify({
"status": "healthy",
"service": "trading-bot-dashboard"
})
if __name__ == '__main__':
print("🚀 Dashboard starting on http://0.0.0.0:5000")
print("📊 Access from other devices on your network")
print("🔒 Basic auth enabled (username: admin, password: admin)")
print("🛑 Press Ctrl+C to stop")
app.run(host='127.0.0.1', port=5000, debug=False)