rob
/
brighter-trading
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
brighter-trading/src/app.py

738 lines
29 KiB
Python
Raw Blame History

# app.py
# Monkey patching must occur before other imports
import eventlet
eventlet.monkey_patch() # noqa: E402
# Standard library imports
import logging # noqa: E402
import os # noqa: E402
import json # noqa: E402
import subprocess # noqa: E402
import xml.etree.ElementTree as ET # noqa: E402
# Third-party imports
from flask import Flask, render_template, request, redirect, jsonify, session, flash # noqa: E402
from flask_cors import CORS # noqa: E402
from flask_socketio import SocketIO, emit, join_room, leave_room, disconnect # noqa: E402
from email_validator import validate_email, EmailNotValidError # noqa: E402
# Local application imports
from BrighterTrades import BrighterTrades # noqa: E402
from utils import sanitize_for_json # noqa: E402
# Set up logging
log_level_name = os.getenv('BRIGHTER_LOG_LEVEL', 'INFO').upper()
log_level = getattr(logging, log_level_name, logging.INFO)
# Debug file logger for execution loop
_loop_debug = logging.getLogger('loop_debug')
_loop_debug.setLevel(logging.DEBUG)
_loop_handler = logging.FileHandler('/home/rob/PycharmProjects/BrighterTrading/trade_debug.log', mode='a')
_loop_handler.setFormatter(logging.Formatter('%(asctime)s - [LOOP] %(message)s'))
_loop_debug.addHandler(_loop_handler)
logging.basicConfig(level=log_level)
logging.getLogger('ccxt.base.exchange').setLevel(logging.WARNING)
logging.getLogger('urllib3').setLevel(logging.WARNING)
# Create a Flask object named app that serves the html.
app = Flask(__name__)
# Create a socket in order to receive requests.
socketio = SocketIO(app, async_mode='eventlet')
# Custom Jinja2 filter to format small balances properly (avoid scientific notation)
@app.template_filter('format_balance')
def format_balance(value):
"""Format balance to avoid scientific notation for small values."""
if value is None:
return '0'
try:
val = float(value)
if val == 0:
return '0'
elif abs(val) < 0.00001:
return f'{val:.8f}'.rstrip('0').rstrip('.')
elif abs(val) < 1:
return f'{val:.6f}'.rstrip('0').rstrip('.')
elif abs(val) < 1000:
return f'{val:.4f}'.rstrip('0').rstrip('.')
else:
return f'{val:,.2f}'
except (ValueError, TypeError):
return str(value)
# Create a BrighterTrades object. This the main application that maintains access to the server, local storage,
# and manages objects that process trade data.
brighter_trades = BrighterTrades(socketio)
# Set server configuration globals.
CORS_HEADERS = 'Content-Type'
# Set the app directly with the globals.
app.config.from_object(__name__)
app.secret_key = '1_BAD_secrete_KEY_is_not_2'
# Enable cross-origin resource sharing for specific endpoints
cors = CORS(app, supports_credentials=True,
resources={
r"/api/indicator_init": {"origins": ["http://127.0.0.1:5000", "http://localhost:5000"]}
},
allow_headers=['Content-Type']) # Change from headers to allow_headers
@app.after_request
def add_cors_headers(response):
response.headers['Access-Control-Allow-Credentials'] = 'true'
return response
# =============================================================================
# Strategy Execution Loop (Background Task)
# =============================================================================
# This runs in the background and periodically executes active live/paper strategies
STRATEGY_LOOP_INTERVAL = 10 # seconds between strategy ticks
_strategy_loop_running = False
def strategy_execution_loop():
"""
Background loop that executes active strategies periodically.
For live trading, strategies need to be triggered server-side,
not just when the client sends candle data.
"""
global _strategy_loop_running
_strategy_loop_running = True
logger = logging.getLogger('strategy_loop')
logger.info("Strategy execution loop started")
while _strategy_loop_running:
try:
# Check if there are any active strategy instances
active_count = brighter_trades.strategies.get_active_count()
# Log every iteration for debugging
if not hasattr(strategy_execution_loop, '_iter_count'):
strategy_execution_loop._iter_count = 0
strategy_execution_loop._iter_count += 1
logger.info(f"Execution loop iteration {strategy_execution_loop._iter_count}, active_count={active_count}")
if active_count > 0:
logger.info(f"Executing {active_count} active strategies...")
# Iterate directly over active instances
instance_keys = list(brighter_trades.strategies.active_instances.keys())
for instance_key in instance_keys:
try:
user_id, strategy_id, mode = instance_key
instance = brighter_trades.strategies.active_instances.get(instance_key)
if instance is None:
continue
# For live strategies, get current price as candle data
# Default to BTC/USDT if no symbol specified
symbol = getattr(instance, 'symbol', 'BTC/USDT')
try:
price = brighter_trades.exchanges.get_price(symbol)
if price:
import time
candle_data = {
'symbol': symbol,
'close': price,
'open': price,
'high': price,
'low': price,
'volume': 0,
'time': int(time.time())
}
# Execute strategy tick
events = instance.tick(candle_data)
if events:
logger.info(f"Strategy {strategy_id} generated {len(events)} events: {events}")
# Emit events to the user's room
user_name = brighter_trades.users.get_username(user_id=user_id)
logger.info(f"Emitting to user_id={user_id}, user_name={user_name}")
if user_name:
socketio.emit('strategy_events', sanitize_for_json({
'strategy_id': strategy_id,
'mode': mode,
'events': events
}), room=user_name)
logger.info(f"Emitted strategy_events to room={user_name}")
else:
logger.warning(f"Could not find username for user_id={user_id}")
except Exception as e:
logger.warning(f"Could not get price for {symbol}: {e}")
except Exception as e:
logger.error(f"Error executing strategy {instance_key}: {e}", exc_info=True)
# Update active trades (runs every iteration, regardless of active strategies)
_loop_debug.debug(f"Checking active_trades: {len(brighter_trades.trades.active_trades)} trades")
if brighter_trades.trades.active_trades:
_loop_debug.debug(f"Has active trades, getting prices...")
try:
# Group trades by (exchange, symbol) for per-exchange price fetching
exchange_symbols = set()
for trade in brighter_trades.trades.active_trades.values():
exchange = getattr(trade, 'exchange', None) or trade.target
# Normalize exchange name (skip test_exchange placeholder)
if exchange and exchange.lower() not in ['test_exchange', 'paper']:
exchange_symbols.add((exchange.lower(), trade.symbol))
else:
# Fall back to default exchange for legacy trades
exchange_symbols.add((None, trade.symbol))
_loop_debug.debug(f"Exchange+symbols to fetch: {exchange_symbols}")
price_updates = {}
for exchange, symbol in exchange_symbols:
try:
price = brighter_trades.exchanges.get_price(symbol, exchange_name=exchange)
_loop_debug.debug(f"Got price for {exchange}:{symbol}: {price}")
if price:
# Use exchange:symbol key for exchange-specific prices
if exchange:
price_updates[f"{exchange}:{symbol}"] = price
else:
price_updates[symbol] = price
except Exception as e:
_loop_debug.debug(f"Failed to get price for {exchange}:{symbol}: {e}")
logger.warning(f"Could not get price for {symbol} from {exchange}: {e}")
_loop_debug.debug(f"price_updates: {price_updates}")
if price_updates:
_loop_debug.debug(f"Calling brighter_trades.trades.update()")
trade_updates = brighter_trades.trades.update(price_updates)
_loop_debug.debug(f"trade_updates returned: {trade_updates}")
if trade_updates:
logger.debug(f"Trade updates (no active strategies): {trade_updates}")
for update in trade_updates:
trade_id = update.get('id')
trade = brighter_trades.trades.active_trades.get(trade_id)
_loop_debug.debug(f"Emitting update for trade_id={trade_id}, creator={trade.creator if trade else None}")
if trade and trade.creator:
user_name = brighter_trades.users.get_username(user_id=trade.creator)
if user_name:
socketio.emit('trade_update', sanitize_for_json(update), room=user_name)
_loop_debug.debug(f"Emitted trade_update to room={user_name}")
except Exception as e:
_loop_debug.debug(f"Exception in trade update: {e}")
logger.error(f"Error updating trades (no strategies): {e}", exc_info=True)
except Exception as e:
logger.error(f"Strategy execution loop error: {e}")
# Sleep before next iteration
eventlet.sleep(STRATEGY_LOOP_INTERVAL)
logger.info("Strategy execution loop stopped")
def start_strategy_loop():
"""Start the strategy execution loop in a background greenlet."""
eventlet.spawn(strategy_execution_loop)
# Start the loop when the app starts (will be called from main block)
def _coerce_user_id(user_id):
if user_id is None or user_id == '':
return None
try:
return int(user_id)
except (TypeError, ValueError):
return None
def resolve_user_name(payload: dict | None) -> str | None:
"""
Resolve a username from payload fields, accepting both legacy and migrated key shapes.
"""
if not isinstance(payload, dict):
return None
user_name = payload.get('user_name') or payload.get('user')
if user_name:
return user_name
user_id = _coerce_user_id(payload.get('user_id') or payload.get('userId'))
if user_id is None:
return None
try:
return brighter_trades.users.get_username(user_id=user_id)
except Exception:
logging.warning(f"Unable to resolve user_name from user id '{user_id}'.")
return None
@app.route('/')
def welcome():
"""
Serves the welcome/landing page.
"""
return render_template('welcome.html')
@app.route('/app')
# @cross_origin(supports_credentials=True)
def index():
"""
Fetches data from brighter_trades and inject it into an HTML template.
Renders the html template and serves the web application.
"""
# session.clear() # only for debugging!!!!
try:
# Log the user in.
user_name = brighter_trades.users.load_or_create_user(username=session.get('user'))
except ValueError as e:
if str(e) != 'GuestLimitExceeded!':
raise
else:
flash('The system is too busy for another user.')
return redirect('/signup')
# Ensure client session is assigned a user_name.
session['user'] = user_name
print('[SERVING INDEX] (USERNAME):', user_name)
default_exchange = 'binance'
default_keys = None # No keys needed for public market data
# Ensure that a valid connection with an exchange exists
result = brighter_trades.connect_user_to_exchange(user_name=user_name,
default_exchange=default_exchange,
default_keys=default_keys)
if not result:
raise ValueError("Couldn't connect to the default exchange.")
# A dict of data required to build the html of the user app.
# Dynamic content like options and titles and balances to display.
rendered_data = brighter_trades.get_rendered_data(user_name)
# Jsom initialization data to be passed into the web application.
js_data = brighter_trades.get_js_init_data(user_name)
# serve the landing page.
return render_template('index.html',
title=rendered_data['title'],
user_name=user_name,
my_balances=rendered_data['my_balances'],
symbols=rendered_data['symbols'],
intervals=rendered_data['intervals'],
interval_state=rendered_data['chart_interval'],
exchanges=rendered_data['available_exchanges'],
cond_exchanges=rendered_data['chart_exchanges'],
confd_exchanges=rendered_data['configured_exchanges'],
selected_exchange=rendered_data['selected_exchange'],
indicator_types=rendered_data['indicator_types'],
indicator_list=rendered_data['indicator_list'],
checked=rendered_data['enabled_indicators'],
ma_vals=rendered_data['ma_vals'],
js_data=js_data,
active_trades=rendered_data['active_trades'],
open_orders=rendered_data['open_orders'])
@socketio.on('connect')
def handle_connect():
user_name = request.args.get('user_name')
if not user_name:
user_name = resolve_user_name({
'userId': request.args.get('userId'),
'user_id': request.args.get('user_id')
})
if user_name and brighter_trades.get_user_info(user_name=user_name, info='Is logged in?'):
# Join a room specific to the user for targeted messaging
room = user_name # You can choose an appropriate room naming strategy
join_room(room)
emit('message', {'reply': 'connected', 'data': 'Connection established'})
else:
emit('message', {'reply': 'error', 'data': 'User not authenticated'})
# Disconnect the client if not authenticated
disconnect()
@socketio.on('message')
def handle_message(data):
"""
Handle incoming JSON messages with authentication.
"""
# Validate input
if 'message_type' not in data or 'data' not in data:
emit('message', {"success": False, "message": "Invalid message format."})
return
msg_type, msg_data = data['message_type'], data['data']
# Extract user_name from the incoming message data
user_name = resolve_user_name(msg_data)
if not user_name:
emit('message', {"success": False, "message": "User not specified"})
return
msg_data.setdefault('user_name', user_name)
try:
user_id = brighter_trades.get_user_info(user_name=user_name, info='User_id')
if user_id is not None:
msg_data.setdefault('user_id', user_id)
msg_data.setdefault('userId', user_id)
except Exception:
pass
# Check if the user is logged in
if not brighter_trades.get_user_info(user_name=user_name, info='Is logged in?'):
emit('message', {"success": False, "message": "User not logged in"})
return
# Process the incoming message based on the type
resp = brighter_trades.process_incoming_message(msg_type=msg_type, msg_data=msg_data, socket_conn_id=request.sid)
# Send the response back to the client
if resp:
emit('message', resp)
@app.route('/settings', methods=['POST'])
def settings():
"""
This route is used to change settings that require a browser refresh.
Options are:
Data source options for the chart display:
timeframe: For setting the time frame the of the viewport.
trading_pair: For setting the market displayed in the viewport.
exchange_name: For setting the exchange_name displayed in the viewport.
Controls of the indicators:
new_indicator: Creates a new indicator and stores it.
edit_indicator: Edits the properties of specific indicators.
toggle_indicator: Enables or disables display of indicators in the viewport.
:return: None - Redirects to index, or returns JSON for async requests.
"""
# Request the user_name from the client session.
if not (user_name := session.get('user')):
return jsonify({"success": False, "message": "Not logged in"}), 401
# Return if the user is not logged in.
if not brighter_trades.get_user_info(user_name=user_name, info='Is logged in?'):
return jsonify({"success": False, "message": "User not logged in"}), 401
# Get the setting that the application wants to change.
try:
# Handle JSON or form submissions
if request.is_json:
data = request.json
setting = data.get('setting')
params = data.get('indicator', {})
else:
setting = request.form.get('setting')
# Use to_dict(flat=False) to get lists for multi-value fields (like checkboxes)
params = request.form.to_dict(flat=False)
# Convert single-value lists back to single values, except for 'indicator'
for key, value in params.items():
if key != 'indicator' and isinstance(value, list) and len(value) == 1:
params[key] = value[0]
if not setting:
return jsonify({"success": False, "message": "No setting provided"}), 400
# Change the setting.
brighter_trades.adjust_setting(user_name=user_name, setting=setting, params=params)
# Return success as JSON if called via an async request
if request.is_json:
return jsonify({"success": True}), 200
# Redirect if this is a form submission (non-async request)
return redirect('/app')
except Exception as e:
return jsonify({"success": False, "message": str(e)}), 500
@app.route('/docs')
def docs():
"""
Documentation page - placeholder for now.
"""
# TODO: Link to actual documentation when available
flash('Documentation coming soon! Redirecting to the app.')
return redirect('/app')
@app.route('/signup')
def signup():
return render_template('sign_up.html', title='title')
@app.route('/signout')
def signout():
if not (user_name := session.get('user')):
return redirect('/')
if brighter_trades.log_user_in_out(user_name=user_name, cmd='logout'):
# If the user was logged out successfully delete the session var.
del session['user']
return redirect('/')
@app.route('/login', methods=['GET'])
def login_page():
"""Display the login page."""
return render_template('login_page.html', title='Login')
@app.route('/login', methods=['POST'])
def login():
# Get the user_name and password from the form data
username = request.form.get('username')
password = request.form.get('password')
# Validate the input
if not username or not password:
flash('Please provide both username and password.', 'login_error')
return redirect('/login')
# Attempt to log in the user
success = brighter_trades.log_user_in_out(user_name=username, cmd='login', password=password)
if success:
# Store the user_name in the session
session['user'] = username
flash('Login successful!', 'success')
return redirect('/app')
else:
flash('Invalid username or password. Please try again.', 'login_error')
return redirect('/login')
@app.route('/signup_submit', methods=['POST'])
def signup_submit():
email = request.form.get('email')
username = request.form.get('user_name')
password = request.form.get('password')
# Validate email format
try:
validate_email(email)
except EmailNotValidError as e:
flash(f"Invalid email format: {e}", 'signup_error')
return redirect('/signup')
# Validate user_name and password
if not username or not password:
flash("Please provide both username and password.", 'signup_error')
return redirect('/signup')
# Create a new user
success = brighter_trades.create_new_user(email=email, username=username, password=password)
if success:
session['user'] = username
flash("Signup successful! You are now logged in.", 'success')
return redirect('/app')
else:
flash("Username or email already exists. Please try a different one.", 'signup_error')
return redirect('/signup')
@app.route('/api/indicator_init', methods=['POST', 'GET'])
def indicator_init():
"""
Initializes the indicators and returns the data for a given symbol and timeframe.
"""
try:
data = request.get_json()
username = resolve_user_name(data)
if not username:
return jsonify({'error': 'Invalid user name.'}), 400
# Check if the user is logged in
if not brighter_trades.get_user_info(user_name=username, info='Is logged in?'):
return jsonify({'error': 'User is not logged in.'}), 401
# Get the chart view for the user
chart_view = brighter_trades.get_user_info(user_name=username, info='Chart View')
# Get the indicator data
source = {'user_name': username, 'market': chart_view}
data = brighter_trades.get_indicator_data(user_name=username, source=source, start_ts=None, num_results=1000)
# indicators={'EMA 5': {'visible': true, 'type': 'EMA', 'color': 'red' },
# 'vol': {'visible': true, 'type': 'Volume'},'New Indicator': {'visible': true, 'type': 'nothing'}}
return jsonify(data), 200
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': str(e)}), 500
@app.route('/api/edm-config', methods=['GET'])
def edm_config():
"""
Returns the EDM (Exchange Data Manager) configuration for the frontend.
This allows the frontend to connect directly to EDM for candle data.
"""
edm_settings = brighter_trades.config.get_setting('edm') or {}
return jsonify({
'rest_url': edm_settings.get('rest_url', 'http://localhost:8080'),
'ws_url': edm_settings.get('ws_url', 'ws://localhost:8080/ws'),
'enabled': edm_settings.get('enabled', True),
}), 200
@app.route('/api/chart-view', methods=['POST'])
def get_chart_view():
"""
Returns the current chart view settings for a user.
Used by the frontend to get exchange, market, and timeframe for EDM requests.
"""
data = request.get_json() or {}
user_name = data.get('user_name')
if not user_name:
return jsonify({'error': 'user_name is required'}), 400
try:
chart_view = brighter_trades.users.get_chart_view(user_name=user_name)
if chart_view:
return jsonify({
'exchange': chart_view.get('exchange', 'binance'),
'market': chart_view.get('market', 'BTC/USDT'),
'timeframe': chart_view.get('timeframe', '1h'),
}), 200
else:
# Return defaults if no chart view is set
return jsonify({
'exchange': 'binance',
'market': 'BTC/USDT',
'timeframe': '1h',
}), 200
except Exception as e:
logging.error(f"Error getting chart view for {user_name}: {e}")
return jsonify({
'exchange': 'binance',
'market': 'BTC/USDT',
'timeframe': '1h',
}), 200
@app.route('/api/generate-strategy', methods=['POST'])
def generate_strategy():
"""
Generate a Blockly strategy from natural language description using AI.
Calls the CmdForge strategy-builder tool.
"""
data = request.get_json() or {}
description = data.get('description', '').strip()
indicators = data.get('indicators', [])
signals = data.get('signals', [])
default_source = data.get('default_source', {
'exchange': 'binance',
'market': 'BTC/USDT',
'timeframe': '5m'
})
if not description:
return jsonify({'error': 'Description is required'}), 400
try:
# Build input for the strategy-builder tool
tool_input = json.dumps({
'description': description,
'indicators': indicators,
'signals': signals,
'default_source': default_source
})
# Call CmdForge strategy-builder tool
result = subprocess.run(
['strategy-builder'],
input=tool_input,
capture_output=True,
text=True,
timeout=120 # 2 minute timeout for AI generation
)
if result.returncode != 0:
error_msg = result.stderr.strip() or 'Strategy generation failed'
logging.error(f"strategy-builder failed: {error_msg}")
return jsonify({'error': error_msg}), 500
workspace_xml = result.stdout.strip()
# Validate the generated XML
if not _validate_blockly_xml(workspace_xml):
logging.error(f"Invalid Blockly XML generated: {workspace_xml[:200]}")
return jsonify({'error': 'Generated strategy is invalid'}), 500
return jsonify({
'success': True,
'workspace_xml': workspace_xml
}), 200
except subprocess.TimeoutExpired:
logging.error("strategy-builder timed out")
return jsonify({'error': 'Strategy generation timed out'}), 504
except FileNotFoundError:
logging.error("strategy-builder tool not found")
return jsonify({'error': 'Strategy builder tool not installed'}), 500
except Exception as e:
logging.error(f"Error generating strategy: {e}")
return jsonify({'error': str(e)}), 500
def _validate_blockly_xml(xml_string: str) -> bool:
"""Validate that the string is valid Blockly XML."""
try:
root = ET.fromstring(xml_string)
# Check it's a Blockly XML document (handle namespace prefixed tags)
# Tag can be 'xml' or '{namespace}xml'
tag_name = root.tag.split('}')[-1] if '}' in root.tag else root.tag
if tag_name != 'xml' and 'blockly' not in root.tag.lower():
return False
# Check it has at least one block using namespace-aware search
# Use .//* to find all descendants, then filter by local name
blocks = [elem for elem in root.iter()
if (elem.tag.split('}')[-1] if '}' in elem.tag else elem.tag) == 'block']
return len(blocks) > 0
except ET.ParseError:
return False
except Exception:
return False
@app.route('/health/edm', methods=['GET'])
def edm_health():
"""
Health check endpoint for EDM connectivity.
"""
if brighter_trades.edm_client is None:
return jsonify({'status': 'disabled', 'message': 'EDM client not initialized'}), 200
try:
healthy = brighter_trades.edm_client.health_check_sync()
if healthy:
return jsonify({'status': 'healthy'}), 200
else:
return jsonify({'status': 'unhealthy', 'message': 'EDM health check failed'}), 503
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)}), 503
if __name__ == '__main__':
# Start the strategy execution loop in the background
start_strategy_loop()
logging.info("Strategy execution loop started in background")
socketio.run(app, host='127.0.0.1', port=5002, debug=False, use_reloader=False)
Reference in New Issue View Git Blame Copy Permalink