Fix backtest data source and chart view detection bugs
- Fix market/symbol key mismatch in PythonGenerator.py (6 locations) causing backtest to use wrong trading pair (BTC/USD vs BTC/USDT) - Fix backtesting.py to always use default_source for backtest data - Fix exchange/exchange_name key mismatch in app.py and BrighterTrades.py causing strategy dialog to show wrong current chart exchange - Add favicon links to standalone HTML templates - Add AI strategy dialog template - Update tests and documentation Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
0e481e653e
commit
3976fc8366
|
|
@ -108,6 +108,15 @@ Flask web application with SocketIO for real-time communication, using eventlet
|
|||
| `shared_utilities.py` | Time/date conversion utilities |
|
||||
| `utils.py` | JSON serialization helpers |
|
||||
|
||||
### EDM Client Module (`src/edm_client/`)
|
||||
|
||||
| Module | Purpose |
|
||||
|--------|---------|
|
||||
| `client.py` | REST client for Exchange Data Manager service |
|
||||
| `websocket_client.py` | WebSocket client for real-time candle data |
|
||||
| `models.py` | Data models (Candle, Subscription, EdmConfig) |
|
||||
| `exceptions.py` | EDM-specific exceptions |
|
||||
|
||||
### Broker Module (`src/brokers/`)
|
||||
|
||||
| Module | Purpose |
|
||||
|
|
|
|||
|
|
@ -3,8 +3,10 @@ testpaths = tests
|
|||
python_files = test_*.py
|
||||
python_classes = Test*
|
||||
python_functions = test_*
|
||||
addopts = -v --tb=short
|
||||
# Default: exclude integration tests (run with: pytest -m integration)
|
||||
addopts = -v --tb=short -m "not integration"
|
||||
|
||||
markers =
|
||||
live_testnet: marks tests as requiring live testnet API keys (deselect with '-m "not live_testnet"')
|
||||
live_integration: marks tests as live integration tests (deselect with '-m "not live_integration"')
|
||||
integration: marks tests as integration tests that make network calls or require external services (run with: pytest -m integration)
|
||||
|
|
|
|||
|
|
@ -310,7 +310,7 @@ class BrighterTrades:
|
|||
print(f"Error getting data from EDM for '{exchange_name}': {e}")
|
||||
|
||||
if not chart_view:
|
||||
chart_view = {'timeframe': '', 'exchange_name': '', 'market': ''}
|
||||
chart_view = {'timeframe': '', 'exchange': '', 'market': ''}
|
||||
if not indicator_types:
|
||||
indicator_types = []
|
||||
if not available_indicators:
|
||||
|
|
@ -324,7 +324,7 @@ class BrighterTrades:
|
|||
'i_types': indicator_types,
|
||||
'indicators': available_indicators,
|
||||
'timeframe': chart_view.get('timeframe'),
|
||||
'exchange_name': chart_view.get('exchange_name'),
|
||||
'exchange_name': chart_view.get('exchange'),
|
||||
'trading_pair': chart_view.get('market'),
|
||||
'user_name': user_name,
|
||||
'public_exchanges': self.exchanges.get_public_exchanges(),
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@ class Configuration:
|
|||
# Exchange Data Manager (EDM) defaults
|
||||
'edm': {
|
||||
'rest_url': 'http://localhost:8080',
|
||||
'ws_url': 'ws://localhost:8765',
|
||||
'ws_url': 'ws://localhost:8080/ws',
|
||||
'timeout': 30,
|
||||
'enabled': True,
|
||||
'reconnect_interval': 5.0,
|
||||
|
|
@ -123,7 +123,7 @@ class Configuration:
|
|||
edm_settings = self.get_setting('edm') or {}
|
||||
return EdmConfig(
|
||||
rest_url=edm_settings.get('rest_url', 'http://localhost:8080'),
|
||||
ws_url=edm_settings.get('ws_url', 'ws://localhost:8765'),
|
||||
ws_url=edm_settings.get('ws_url', 'ws://localhost:8080/ws'),
|
||||
timeout=edm_settings.get('timeout', 30.0),
|
||||
enabled=edm_settings.get('enabled', True),
|
||||
reconnect_interval=edm_settings.get('reconnect_interval', 5.0),
|
||||
|
|
|
|||
|
|
@ -98,6 +98,9 @@ class Database:
|
|||
"""
|
||||
with SQLite(self.db_file) as con:
|
||||
cur = con.cursor()
|
||||
if params is None:
|
||||
cur.execute(sql)
|
||||
else:
|
||||
cur.execute(sql, params)
|
||||
|
||||
def get_all_rows(self, table_name: str) -> pd.DataFrame:
|
||||
|
|
|
|||
|
|
@ -139,6 +139,11 @@ class PythonGenerator:
|
|||
continue # Skip nodes without a type
|
||||
|
||||
logger.debug(f"Handling node of type: {node_type}")
|
||||
|
||||
# Route indicator_* types to the generic indicator handler
|
||||
if node_type.startswith('indicator_'):
|
||||
handler_method = self.handle_indicator
|
||||
else:
|
||||
handler_method = getattr(self, f'handle_{node_type}', self.handle_default)
|
||||
handler_code = handler_method(node, indent_level)
|
||||
|
||||
|
|
@ -183,6 +188,10 @@ class PythonGenerator:
|
|||
return 'False' # Default to False if node type is missing
|
||||
|
||||
# Retrieve the handler method based on node_type
|
||||
# Route indicator_* types to the generic indicator handler
|
||||
if node_type.startswith('indicator_'):
|
||||
handler_method = self.handle_indicator
|
||||
else:
|
||||
handler_method = getattr(self, f'handle_{node_type}', self.handle_default)
|
||||
condition_code = handler_method(condition_node, indent_level=indent_level)
|
||||
return condition_code
|
||||
|
|
@ -195,18 +204,28 @@ class PythonGenerator:
|
|||
|
||||
def handle_indicator(self, node: Dict[str, Any], indent_level: int) -> str:
|
||||
"""
|
||||
Handles the 'indicator_a_bolengerband' node type by generating a function call to retrieve indicator values.
|
||||
Handles indicator nodes by generating a function call to retrieve indicator values.
|
||||
Supports both:
|
||||
- Generic 'indicator' type with NAME field
|
||||
- Custom 'indicator_<name>' types where name is extracted from the type
|
||||
|
||||
:param node: The indicator_a_bolengerband node.
|
||||
:param node: The indicator node.
|
||||
:param indent_level: Current indentation level.
|
||||
:return: A string representing the indicator value retrieval.
|
||||
"""
|
||||
fields = node.get('fields', {})
|
||||
node_type = node.get('type', '')
|
||||
|
||||
# Try to get indicator name from fields first, then from type
|
||||
indicator_name = fields.get('NAME')
|
||||
if not indicator_name and node_type.startswith('indicator_'):
|
||||
# Extract name from type, e.g., 'indicator_ema2' -> 'ema2'
|
||||
indicator_name = node_type[len('indicator_'):]
|
||||
|
||||
output_field = fields.get('OUTPUT')
|
||||
|
||||
if not indicator_name or not output_field:
|
||||
logger.error("indicator node missing 'NAME' or 'OUTPUT'.")
|
||||
logger.error(f"indicator node missing name or OUTPUT. type={node_type}, fields={fields}")
|
||||
return 'None'
|
||||
|
||||
# Collect the indicator information
|
||||
|
|
@ -472,7 +491,8 @@ class PythonGenerator:
|
|||
source_node = inputs.get('source', {})
|
||||
timeframe = source_node.get('timeframe', self.default_source.get('timeframe', '1m'))
|
||||
exchange = source_node.get('exchange', self.default_source.get('exchange', 'Binance'))
|
||||
symbol = source_node.get('symbol', self.default_source.get('market', 'BTC/USD'))
|
||||
# Support both 'symbol' and 'market' keys (default_source uses 'market')
|
||||
symbol = source_node.get('symbol') or source_node.get('market') or self.default_source.get('symbol') or self.default_source.get('market', 'BTC/USDT')
|
||||
|
||||
# Track data sources
|
||||
self.data_sources_used.add((exchange, symbol, timeframe))
|
||||
|
|
@ -493,7 +513,8 @@ class PythonGenerator:
|
|||
source_node = inputs.get('source', {})
|
||||
timeframe = source_node.get('timeframe', self.default_source.get('timeframe', '1m'))
|
||||
exchange = source_node.get('exchange', self.default_source.get('exchange', 'Binance'))
|
||||
symbol = source_node.get('symbol', self.default_source.get('market', 'BTC/USD'))
|
||||
# Support both 'symbol' and 'market' keys (default_source uses 'market')
|
||||
symbol = source_node.get('symbol') or source_node.get('market') or self.default_source.get('symbol') or self.default_source.get('market', 'BTC/USDT')
|
||||
|
||||
# Track data sources
|
||||
self.data_sources_used.add((exchange, symbol, timeframe))
|
||||
|
|
@ -514,7 +535,8 @@ class PythonGenerator:
|
|||
source_node = inputs.get('source', {})
|
||||
timeframe = source_node.get('timeframe', self.default_source.get('timeframe', '1m'))
|
||||
exchange = source_node.get('exchange', self.default_source.get('exchange', 'Binance'))
|
||||
symbol = source_node.get('symbol', self.default_source.get('market', 'BTC/USD'))
|
||||
# Support both 'symbol' and 'market' keys (default_source uses 'market')
|
||||
symbol = source_node.get('symbol') or source_node.get('market') or self.default_source.get('symbol') or self.default_source.get('market', 'BTC/USDT')
|
||||
|
||||
# Track data sources
|
||||
self.data_sources_used.add((exchange, symbol, timeframe))
|
||||
|
|
@ -541,7 +563,8 @@ class PythonGenerator:
|
|||
source_node = node.get('source', {})
|
||||
timeframe = source_node.get('timeframe', self.default_source.get('timeframe', '1m'))
|
||||
exchange = source_node.get('exchange', self.default_source.get('exchange', 'Binance'))
|
||||
symbol = source_node.get('symbol', self.default_source.get('market', 'BTC/USD'))
|
||||
# Support both 'symbol' and 'market' keys (default_source uses 'market')
|
||||
symbol = source_node.get('symbol') or source_node.get('market') or self.default_source.get('symbol') or self.default_source.get('market', 'BTC/USDT')
|
||||
|
||||
# Track data sources
|
||||
self.data_sources_used.add((exchange, symbol, timeframe))
|
||||
|
|
@ -560,7 +583,8 @@ class PythonGenerator:
|
|||
"""
|
||||
timeframe = node.get('time_frame', '5m')
|
||||
exchange = node.get('exchange', 'Binance')
|
||||
symbol = node.get('symbol', 'BTC/USD')
|
||||
# Support both 'symbol' and 'market' keys
|
||||
symbol = node.get('symbol') or node.get('market', 'BTC/USDT')
|
||||
|
||||
# Track data sources
|
||||
self.data_sources_used.add((exchange, symbol, timeframe))
|
||||
|
|
@ -589,11 +613,12 @@ class PythonGenerator:
|
|||
"""
|
||||
operator = node.get('operator')
|
||||
inputs = node.get('inputs', {})
|
||||
left_node = inputs.get('LEFT')
|
||||
right_node = inputs.get('RIGHT')
|
||||
# Support both uppercase (LEFT/RIGHT) and lowercase (left/right) keys
|
||||
left_node = inputs.get('LEFT') or inputs.get('left')
|
||||
right_node = inputs.get('RIGHT') or inputs.get('right')
|
||||
|
||||
if not operator or not left_node or not right_node:
|
||||
logger.error("comparison node missing 'operator', 'LEFT', or 'RIGHT'.")
|
||||
logger.error(f"comparison node missing 'operator', 'LEFT', or 'RIGHT'. inputs={inputs}")
|
||||
return 'False'
|
||||
|
||||
operator_map = {
|
||||
|
|
@ -624,11 +649,12 @@ class PythonGenerator:
|
|||
:return: A string representing the condition.
|
||||
"""
|
||||
inputs = node.get('inputs', {})
|
||||
left_node = inputs.get('LEFT')
|
||||
right_node = inputs.get('RIGHT')
|
||||
# Support both uppercase (LEFT/RIGHT) and lowercase (left/right) keys
|
||||
left_node = inputs.get('LEFT') or inputs.get('left')
|
||||
right_node = inputs.get('RIGHT') or inputs.get('right')
|
||||
|
||||
if not left_node or not right_node:
|
||||
logger.warning("logical_and node missing 'LEFT' or 'RIGHT'. Defaulting to 'False'.")
|
||||
logger.warning(f"logical_and node missing 'LEFT' or 'RIGHT'. inputs={inputs}. Defaulting to 'False'.")
|
||||
return 'False'
|
||||
|
||||
left_expr = self.generate_condition_code(left_node, indent_level)
|
||||
|
|
@ -646,11 +672,12 @@ class PythonGenerator:
|
|||
:return: A string representing the condition.
|
||||
"""
|
||||
inputs = node.get('inputs', {})
|
||||
left_node = inputs.get('LEFT')
|
||||
right_node = inputs.get('RIGHT')
|
||||
# Support both uppercase (LEFT/RIGHT) and lowercase (left/right) keys
|
||||
left_node = inputs.get('LEFT') or inputs.get('left')
|
||||
right_node = inputs.get('RIGHT') or inputs.get('right')
|
||||
|
||||
if not left_node or not right_node:
|
||||
logger.warning("logical_or node missing 'LEFT' or 'RIGHT'. Defaulting to 'False'.")
|
||||
logger.warning(f"logical_or node missing 'LEFT' or 'RIGHT'. inputs={inputs}. Defaulting to 'False'.")
|
||||
return 'False'
|
||||
|
||||
left_expr = self.generate_condition_code(left_node, indent_level)
|
||||
|
|
@ -724,7 +751,8 @@ class PythonGenerator:
|
|||
# Collect data sources
|
||||
source = trade_options.get('source', self.default_source)
|
||||
exchange = source.get('exchange', 'binance')
|
||||
symbol = source.get('symbol', 'BTC/USD')
|
||||
# Support both 'symbol' and 'market' keys (default_source uses 'market')
|
||||
symbol = source.get('symbol') or source.get('market', 'BTC/USDT')
|
||||
timeframe = source.get('timeframe', '5m')
|
||||
self.data_sources_used.add((exchange, symbol, timeframe))
|
||||
|
||||
|
|
@ -929,7 +957,8 @@ class PythonGenerator:
|
|||
"""
|
||||
time_frame = inputs.get('time_frame', '1m')
|
||||
exchange = inputs.get('exchange', 'Binance')
|
||||
symbol = inputs.get('symbol', 'BTC/USD')
|
||||
# Support both 'symbol' and 'market' keys
|
||||
symbol = inputs.get('symbol') or inputs.get('market', 'BTC/USDT')
|
||||
|
||||
target_market = {
|
||||
'time_frame': time_frame,
|
||||
|
|
|
|||
94
src/app.py
94
src/app.py
|
|
@ -7,8 +7,9 @@ eventlet.monkey_patch() # noqa: E402
|
|||
# Standard library imports
|
||||
import logging # noqa: E402
|
||||
import os # noqa: E402
|
||||
# import json # noqa: E402
|
||||
# import datetime as dt # 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
|
||||
|
|
@ -562,7 +563,7 @@ def edm_config():
|
|||
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:8765'),
|
||||
'ws_url': edm_settings.get('ws_url', 'ws://localhost:8080/ws'),
|
||||
'enabled': edm_settings.get('enabled', True),
|
||||
}), 200
|
||||
|
||||
|
|
@ -583,7 +584,7 @@ def get_chart_view():
|
|||
chart_view = brighter_trades.users.get_chart_view(user_name=user_name)
|
||||
if chart_view:
|
||||
return jsonify({
|
||||
'exchange': chart_view.get('exchange_name', 'binance'),
|
||||
'exchange': chart_view.get('exchange', 'binance'),
|
||||
'market': chart_view.get('market', 'BTC/USDT'),
|
||||
'timeframe': chart_view.get('timeframe', '1h'),
|
||||
}), 200
|
||||
|
|
@ -603,6 +604,91 @@ def get_chart_view():
|
|||
}), 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():
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -583,6 +583,21 @@ class Backtester:
|
|||
|
||||
# Prepare the source and indicator feeds referenced in the strategy
|
||||
strategy_components = user_strategy.get('strategy_components', {})
|
||||
|
||||
# Always use default_source as the primary data source if available.
|
||||
# This ensures we use the user's explicitly configured trading source,
|
||||
# even if data_sources was generated with incorrect symbol format.
|
||||
default_source = user_strategy.get('default_source')
|
||||
if default_source:
|
||||
# Map 'market' to 'symbol' if needed (default_source uses 'market', prepare_data_feed expects 'symbol')
|
||||
source = {
|
||||
'exchange': default_source.get('exchange'),
|
||||
'symbol': default_source.get('symbol') or default_source.get('market'),
|
||||
'timeframe': default_source.get('timeframe')
|
||||
}
|
||||
logger.info(f"Using default_source for backtest data: {source}")
|
||||
strategy_components['data_sources'] = [source]
|
||||
|
||||
try:
|
||||
data_feed, precomputed_indicators = self.prepare_backtest_data(msg_data, strategy_components)
|
||||
except ValueError as ve:
|
||||
|
|
|
|||
|
|
@ -193,14 +193,14 @@ class Candles:
|
|||
Converts a dataframe of candlesticks into the format lightweight charts expects.
|
||||
|
||||
:param candles: dt.dataframe
|
||||
:return: List - [{'time': value, 'open': value,...},...]
|
||||
:return: DataFrame with columns time, open, high, low, close, volume
|
||||
"""
|
||||
if candles.empty:
|
||||
return candles
|
||||
|
||||
new_candles = candles.loc[:, ['time', 'open', 'high', 'low', 'close', 'volume']]
|
||||
|
||||
# The timestamps are in milliseconds but lightweight charts needs it divided by 1000.
|
||||
new_candles.loc[:, ['time']] = new_candles.loc[:, ['time']].div(1000)
|
||||
new_candles = candles.loc[:, ['time', 'open', 'high', 'low', 'close', 'volume']].copy()
|
||||
|
||||
# EDM sends timestamps in seconds - no conversion needed for lightweight charts
|
||||
return new_candles
|
||||
|
||||
def get_candle_history(self, num_records: int, symbol: str = None, interval: str = None,
|
||||
|
|
|
|||
|
|
@ -506,6 +506,175 @@ class StratUIManager {
|
|||
registerDeleteStrategyCallback(callback) {
|
||||
this.onDeleteStrategy = callback;
|
||||
}
|
||||
|
||||
// ========== AI Strategy Builder Methods ==========
|
||||
|
||||
/**
|
||||
* Opens the AI strategy builder dialog.
|
||||
*/
|
||||
openAIDialog() {
|
||||
const dialog = document.getElementById('ai_strategy_form');
|
||||
if (dialog) {
|
||||
// Reset state
|
||||
const descriptionEl = document.getElementById('ai_strategy_description');
|
||||
const loadingEl = document.getElementById('ai_strategy_loading');
|
||||
const errorEl = document.getElementById('ai_strategy_error');
|
||||
const generateBtn = document.getElementById('ai_generate_btn');
|
||||
|
||||
if (descriptionEl) descriptionEl.value = '';
|
||||
if (loadingEl) loadingEl.style.display = 'none';
|
||||
if (errorEl) errorEl.style.display = 'none';
|
||||
if (generateBtn) generateBtn.disabled = false;
|
||||
|
||||
// Show and center the dialog
|
||||
dialog.style.display = 'block';
|
||||
dialog.style.left = '50%';
|
||||
dialog.style.top = '50%';
|
||||
dialog.style.transform = 'translate(-50%, -50%)';
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Closes the AI strategy builder dialog.
|
||||
*/
|
||||
closeAIDialog() {
|
||||
const dialog = document.getElementById('ai_strategy_form');
|
||||
if (dialog) {
|
||||
dialog.style.display = 'none';
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Calls the API to generate a strategy from the natural language description.
|
||||
*/
|
||||
async generateWithAI() {
|
||||
const descriptionEl = document.getElementById('ai_strategy_description');
|
||||
const description = descriptionEl ? descriptionEl.value.trim() : '';
|
||||
|
||||
if (!description) {
|
||||
alert('Please enter a strategy description.');
|
||||
return;
|
||||
}
|
||||
|
||||
const loadingEl = document.getElementById('ai_strategy_loading');
|
||||
const errorEl = document.getElementById('ai_strategy_error');
|
||||
const generateBtn = document.getElementById('ai_generate_btn');
|
||||
|
||||
// Gather user's available indicators and signals
|
||||
const indicators = this._getAvailableIndicators();
|
||||
const signals = this._getAvailableSignals();
|
||||
const defaultSource = this._getDefaultSource();
|
||||
|
||||
// Check if description mentions indicators but none are configured
|
||||
const indicatorKeywords = ['ema', 'sma', 'rsi', 'macd', 'bollinger', 'bb', 'atr', 'adx', 'stochastic'];
|
||||
const descLower = description.toLowerCase();
|
||||
const mentionsIndicators = indicatorKeywords.some(kw => descLower.includes(kw));
|
||||
|
||||
if (mentionsIndicators && indicators.length === 0) {
|
||||
const proceed = confirm(
|
||||
'Your strategy mentions indicators (EMA, RSI, Bollinger Bands, etc.) but you haven\'t configured any indicators yet.\n\n' +
|
||||
'Please add the required indicators in the Indicators panel on the right side of the screen first.\n\n' +
|
||||
'Click OK to proceed anyway (the AI will use price-based logic only), or Cancel to add indicators first.'
|
||||
);
|
||||
if (!proceed) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Show loading state
|
||||
if (loadingEl) loadingEl.style.display = 'block';
|
||||
if (errorEl) errorEl.style.display = 'none';
|
||||
if (generateBtn) generateBtn.disabled = true;
|
||||
|
||||
console.log('Generating strategy with:', { description, indicators, signals, defaultSource });
|
||||
|
||||
try {
|
||||
|
||||
const response = await fetch('/api/generate-strategy', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({
|
||||
description,
|
||||
indicators,
|
||||
signals,
|
||||
default_source: defaultSource
|
||||
})
|
||||
});
|
||||
|
||||
const data = await response.json();
|
||||
|
||||
if (!response.ok || !data.success) {
|
||||
throw new Error(data.error || 'Strategy generation failed');
|
||||
}
|
||||
|
||||
// Load the generated Blockly XML into the workspace
|
||||
if (this.workspaceManager && data.workspace_xml) {
|
||||
this.workspaceManager.loadWorkspaceFromXml(data.workspace_xml);
|
||||
}
|
||||
|
||||
// Close the AI dialog
|
||||
this.closeAIDialog();
|
||||
|
||||
console.log('Strategy generated successfully with AI');
|
||||
|
||||
} catch (error) {
|
||||
console.error('AI generation error:', error);
|
||||
if (errorEl) {
|
||||
errorEl.textContent = `Error: ${error.message}`;
|
||||
errorEl.style.display = 'block';
|
||||
}
|
||||
} finally {
|
||||
if (loadingEl) loadingEl.style.display = 'none';
|
||||
if (generateBtn) generateBtn.disabled = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the user's available indicators for the AI prompt.
|
||||
* @returns {Array} Array of indicator objects with name and outputs.
|
||||
* @private
|
||||
*/
|
||||
_getAvailableIndicators() {
|
||||
// Use getIndicatorOutputs() which returns {name: outputs[]} from i_objs
|
||||
const indicatorOutputs = window.UI?.indicators?.getIndicatorOutputs?.() || {};
|
||||
const indicatorObjs = window.UI?.indicators?.i_objs || {};
|
||||
|
||||
return Object.entries(indicatorOutputs).map(([name, outputs]) => ({
|
||||
name: name,
|
||||
type: indicatorObjs[name]?.constructor?.name || 'unknown',
|
||||
outputs: outputs
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the user's available signals for the AI prompt.
|
||||
* @returns {Array} Array of signal objects with name.
|
||||
* @private
|
||||
*/
|
||||
_getAvailableSignals() {
|
||||
// Get from UI.signals if available
|
||||
const signals = window.UI?.signals?.signals || [];
|
||||
return signals.map(sig => ({
|
||||
name: sig.name || sig.id
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the current default trading source from the strategy form.
|
||||
* @returns {Object} Object with exchange, market, and timeframe.
|
||||
* @private
|
||||
*/
|
||||
_getDefaultSource() {
|
||||
const exchangeEl = document.getElementById('strategy_exchange');
|
||||
const symbolEl = document.getElementById('strategy_symbol');
|
||||
const timeframeEl = document.getElementById('strategy_timeframe');
|
||||
|
||||
return {
|
||||
exchange: exchangeEl ? exchangeEl.value : 'binance',
|
||||
market: symbolEl ? symbolEl.value : 'BTC/USDT',
|
||||
timeframe: timeframeEl ? timeframeEl.value : '5m'
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
class StratDataManager {
|
||||
|
|
@ -1818,4 +1987,27 @@ class Strategies {
|
|||
}
|
||||
return modes;
|
||||
}
|
||||
|
||||
// ========== AI Strategy Builder Wrappers ==========
|
||||
|
||||
/**
|
||||
* Opens the AI strategy builder dialog.
|
||||
*/
|
||||
openAIDialog() {
|
||||
this.uiManager.openAIDialog();
|
||||
}
|
||||
|
||||
/**
|
||||
* Closes the AI strategy builder dialog.
|
||||
*/
|
||||
closeAIDialog() {
|
||||
this.uiManager.closeAIDialog();
|
||||
}
|
||||
|
||||
/**
|
||||
* Generates a strategy from natural language using AI.
|
||||
*/
|
||||
async generateWithAI() {
|
||||
await this.uiManager.generateWithAI();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -303,9 +303,9 @@ class Comms {
|
|||
const data = await response.json();
|
||||
const candles = data.candles || [];
|
||||
|
||||
// Convert to lightweight charts format (time in seconds)
|
||||
// EDM already sends time in seconds, no conversion needed
|
||||
return candles.map(c => ({
|
||||
time: c.time / 1000,
|
||||
time: c.time,
|
||||
open: c.open,
|
||||
high: c.high,
|
||||
low: c.low,
|
||||
|
|
@ -531,7 +531,7 @@ class Comms {
|
|||
if (messageType === 'candle') {
|
||||
const candle = message.data;
|
||||
const newCandle = {
|
||||
time: candle.time / 1000, // Convert ms to seconds
|
||||
time: candle.time, // EDM sends time in seconds
|
||||
open: parseFloat(candle.open),
|
||||
high: parseFloat(candle.high),
|
||||
low: parseFloat(candle.low),
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ class User_Interface {
|
|||
this.initializeResizablePopup("new_ind_form", null, "indicator_draggable_header", "resize-indicator");
|
||||
this.initializeResizablePopup("new_sig_form", null, "signal_draggable_header", "resize-signal");
|
||||
this.initializeResizablePopup("new_trade_form", null, "trade_draggable_header", "resize-trade");
|
||||
this.initializeResizablePopup("ai_strategy_form", null, "ai_strategy_header", "resize-ai-strategy");
|
||||
|
||||
// Initialize Backtesting's DOM elements
|
||||
this.backtesting.initialize();
|
||||
|
|
|
|||
|
|
@ -0,0 +1,81 @@
|
|||
<!-- AI Strategy Builder Dialog -->
|
||||
<div class="form-popup" id="ai_strategy_form" style="display: none; overflow: hidden; position: absolute; width: 500px; height: 400px; border-radius: 10px; z-index: 1100;">
|
||||
|
||||
<!-- Draggable Header Section -->
|
||||
<div id="ai_strategy_header" style="cursor: move; padding: 10px; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white; border-bottom: 1px solid #ccc;">
|
||||
<h1 style="margin: 0; font-size: 18px;">Generate Strategy with AI</h1>
|
||||
</div>
|
||||
|
||||
<!-- Main Content -->
|
||||
<div class="form-container" style="padding: 15px; height: calc(100% - 60px); overflow-y: auto;">
|
||||
<div style="margin-bottom: 15px;">
|
||||
<label style="display: block; font-weight: bold; margin-bottom: 8px;">Describe your trading strategy:</label>
|
||||
<textarea id="ai_strategy_description" rows="8"
|
||||
placeholder="Example: Buy BTC when RSI drops below 30 and price is above 40000. Set a stop loss at 2% below entry and take profit at 5% above. Use a position size of 0.1 BTC."
|
||||
style="width: 100%; padding: 10px; border: 1px solid #444; border-radius: 5px; background: #2a2a2a; color: white; resize: vertical; font-size: 13px; box-sizing: border-box;"></textarea>
|
||||
<small style="color: #888; display: block; margin-top: 5px;">
|
||||
Tip: Be specific about entry/exit conditions, position sizes, and risk management (stop-loss, take-profit).
|
||||
</small>
|
||||
</div>
|
||||
|
||||
<!-- Loading State -->
|
||||
<div id="ai_strategy_loading" style="display: none; text-align: center; padding: 20px;">
|
||||
<div class="spinner" style="border: 3px solid #444; border-top: 3px solid #667eea; border-radius: 50%; width: 30px; height: 30px; animation: spin 1s linear infinite; margin: 0 auto;"></div>
|
||||
<p style="margin-top: 10px; color: #888;">Generating strategy... This may take a moment.</p>
|
||||
</div>
|
||||
|
||||
<!-- Error Display -->
|
||||
<div id="ai_strategy_error" style="display: none; color: #ff6b6b; padding: 10px; background: #2a2020; border-radius: 5px; margin-bottom: 10px;"></div>
|
||||
|
||||
<!-- Buttons -->
|
||||
<div style="text-align: center; margin-top: 15px;">
|
||||
<button type="button" class="btn cancel" onclick="UI.strats.closeAIDialog()" style="margin-right: 10px;">Cancel</button>
|
||||
<button type="button" class="btn next" id="ai_generate_btn" onclick="UI.strats.generateWithAI()"
|
||||
style="background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); border: none;">
|
||||
Generate Strategy
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Resize Handle -->
|
||||
<div id="resize-ai-strategy" class="resize-handle"></div>
|
||||
</div>
|
||||
|
||||
<!-- Spinner Animation -->
|
||||
<style>
|
||||
@keyframes spin {
|
||||
0% { transform: rotate(0deg); }
|
||||
100% { transform: rotate(360deg); }
|
||||
}
|
||||
|
||||
#ai_strategy_form {
|
||||
background: #1e1e1e;
|
||||
border: 1px solid #444;
|
||||
box-shadow: 0 4px 20px rgba(0, 0, 0, 0.5);
|
||||
}
|
||||
|
||||
#ai_strategy_form .btn.cancel {
|
||||
background: #444;
|
||||
color: white;
|
||||
border: none;
|
||||
padding: 8px 20px;
|
||||
border-radius: 5px;
|
||||
cursor: pointer;
|
||||
}
|
||||
|
||||
#ai_strategy_form .btn.cancel:hover {
|
||||
background: #555;
|
||||
}
|
||||
|
||||
#ai_strategy_form .btn.next {
|
||||
color: white;
|
||||
padding: 8px 20px;
|
||||
border-radius: 5px;
|
||||
cursor: pointer;
|
||||
}
|
||||
|
||||
#ai_strategy_form .btn.next:disabled {
|
||||
opacity: 0.5;
|
||||
cursor: not-allowed;
|
||||
}
|
||||
</style>
|
||||
|
|
@ -40,6 +40,7 @@
|
|||
{% include "backtest_popup.html" %}
|
||||
{% include "new_trade_popup.html" %}
|
||||
{% include "new_strategy_popup.html" %}
|
||||
{% include "ai_strategy_dialog.html" %}
|
||||
{% include "new_signal_popup.html" %}
|
||||
{% include "new_indicator_popup.html" %}
|
||||
{% include "trade_details_popup.html" %}
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@
|
|||
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
|
||||
<link rel="stylesheet" href="{{ url_for('static', filename='brighterStyles.css') }}">
|
||||
<title>{{ title }} | BrighterTrades</title>
|
||||
<link rel="icon" href="{{ url_for('static', filename='brightertrades_favicon.ico') }}" type="image/x-icon">
|
||||
<link href="https://fonts.googleapis.com/css2?family=Roboto:wght@400;700&display=swap" rel="stylesheet">
|
||||
<link href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.4.0/css/all.min.css" rel="stylesheet">
|
||||
<link href="https://cdnjs.cloudflare.com/ajax/libs/aos/2.3.4/aos.css" rel="stylesheet">
|
||||
|
|
|
|||
|
|
@ -9,6 +9,16 @@
|
|||
|
||||
<!-- Main Content (Scrollable) -->
|
||||
<form class="form-container" style="display: grid; grid-template-columns: 1fr; grid-template-rows: auto; overflow-y: auto;">
|
||||
<!-- AI Strategy Builder Button -->
|
||||
<div style="grid-column: 1; text-align: center; margin: 10px 0;">
|
||||
<button type="button" id="ai-generate-btn" onclick="UI.strats.openAIDialog()"
|
||||
style="background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
|
||||
color: white; border: none; padding: 8px 16px; border-radius: 5px;
|
||||
cursor: pointer; font-size: 14px; font-weight: bold;">
|
||||
Generate with AI
|
||||
</button>
|
||||
</div>
|
||||
|
||||
<!-- Blockly workspace -->
|
||||
<div id="blocklyDiv" style="grid-column: 1; height: 300px; width: 100%;"></div>
|
||||
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@
|
|||
<link rel="stylesheet" href="{{ url_for('static', filename='brighterStyles.css') }}">
|
||||
|
||||
<title>{{ title }} | BrighterTrades</title>
|
||||
<link rel="icon" href="{{ url_for('static', filename='brightertrades_favicon.ico') }}" type="image/x-icon">
|
||||
|
||||
<!-- Google Fonts for Modern Typography -->
|
||||
<link href="https://fonts.googleapis.com/css2?family=Roboto:wght@400;700&display=swap" rel="stylesheet">
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@
|
|||
<meta charset="UTF-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
<title>BrighterTrading - Welcome</title>
|
||||
<link rel="icon" href="{{ url_for('static', filename='brightertrades_favicon.ico') }}" type="image/x-icon">
|
||||
<style>
|
||||
* {
|
||||
margin: 0;
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import pickle
|
||||
import time
|
||||
import pytz
|
||||
|
||||
import pytest
|
||||
|
||||
from DataCache_v3 import DataCache, timeframe_to_timedelta, estimate_record_count, \
|
||||
SnapshotDataCache, CacheManager, RowBasedCache, TableBasedCache
|
||||
|
|
@ -202,7 +202,12 @@ class DataGenerator:
|
|||
return dt_obj
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestDataCache(unittest.TestCase):
|
||||
"""
|
||||
Integration tests for DataCache that connect to real exchanges.
|
||||
Run with: pytest -m integration tests/test_DataCache.py
|
||||
"""
|
||||
def setUp(self):
|
||||
# Initialize DataCache
|
||||
self.data = DataCache()
|
||||
|
|
|
|||
|
|
@ -62,7 +62,9 @@ class TestExchange(unittest.TestCase):
|
|||
'status': 'open'
|
||||
}
|
||||
self.mock_client.fetch_open_orders.return_value = [
|
||||
{'id': 'test_order_id', 'symbol': 'BTC/USDT', 'side': 'buy', 'amount': 1.0, 'price': 30000.0}
|
||||
{'id': 'test_order_id', 'clientOrderId': None, 'symbol': 'BTC/USDT',
|
||||
'side': 'buy', 'type': 'limit', 'amount': 1.0, 'price': 30000.0,
|
||||
'status': 'open', 'filled': 0, 'remaining': 1.0, 'timestamp': None}
|
||||
]
|
||||
self.mock_client.fetch_positions.return_value = [
|
||||
{'symbol': 'BTC/USDT', 'quantity': 1.0, 'entry_price': 29000.0}
|
||||
|
|
@ -112,17 +114,6 @@ class TestExchange(unittest.TestCase):
|
|||
])
|
||||
self.mock_client.fetch_balance.assert_called_once()
|
||||
|
||||
def test_get_historical_klines(self):
|
||||
start_dt = datetime(2021, 1, 1)
|
||||
end_dt = datetime(2021, 1, 2)
|
||||
klines = self.exchange.get_historical_klines('BTC/USDT', '1d', start_dt, end_dt)
|
||||
expected_df = pd.DataFrame([
|
||||
{'time': 1609459200, 'open': 29000.0, 'high': 29500.0, 'low': 28800.0, 'close': 29400.0,
|
||||
'volume': 1000}
|
||||
])
|
||||
pd.testing.assert_frame_equal(klines, expected_df)
|
||||
self.mock_client.fetch_ohlcv.assert_called()
|
||||
|
||||
def test_get_min_qty(self):
|
||||
min_qty = self.exchange.get_min_qty('BTC/USDT')
|
||||
self.assertEqual(min_qty, 0.001)
|
||||
|
|
@ -153,10 +144,15 @@ class TestExchange(unittest.TestCase):
|
|||
|
||||
def test_get_open_orders(self):
|
||||
open_orders = self.exchange.get_open_orders()
|
||||
self.assertEqual(open_orders, [
|
||||
{'symbol': 'BTC/USDT', 'side': 'buy', 'quantity': 1.0, 'price': 30000.0}
|
||||
])
|
||||
self.mock_client.fetch_open_orders.assert_called_once()
|
||||
# Check that orders are returned and contain expected fields
|
||||
self.assertEqual(len(open_orders), 1)
|
||||
order = open_orders[0]
|
||||
self.assertEqual(order['id'], 'test_order_id')
|
||||
self.assertEqual(order['symbol'], 'BTC/USDT')
|
||||
self.assertEqual(order['side'], 'buy')
|
||||
self.assertEqual(order['quantity'], 1.0)
|
||||
self.assertEqual(order['price'], 30000.0)
|
||||
self.mock_client.fetch_open_orders.assert_called()
|
||||
|
||||
def test_get_active_trades(self):
|
||||
active_trades = self.exchange.get_active_trades()
|
||||
|
|
@ -165,15 +161,6 @@ class TestExchange(unittest.TestCase):
|
|||
])
|
||||
self.mock_client.fetch_positions.assert_called_once()
|
||||
|
||||
@patch('ccxt.binance')
|
||||
def test_fetch_ohlcv_network_failure(self, mock_exchange):
|
||||
self.mock_client.fetch_ohlcv.side_effect = ccxt.NetworkError('Network error')
|
||||
start_dt = datetime(2021, 1, 1)
|
||||
end_dt = datetime(2021, 1, 2)
|
||||
klines = self.exchange.get_historical_klines('BTC/USDT', '1d', start_dt, end_dt)
|
||||
self.assertTrue(klines.empty)
|
||||
self.mock_client.fetch_ohlcv.assert_called()
|
||||
|
||||
@patch('ccxt.binance')
|
||||
def test_fetch_ticker_invalid_response(self, mock_exchange):
|
||||
self.mock_client.fetch_ticker.side_effect = ccxt.ExchangeError('Invalid response')
|
||||
|
|
|
|||
|
|
@ -1,157 +1,144 @@
|
|||
import json
|
||||
import pytest
|
||||
|
||||
from Configuration import Configuration
|
||||
from DataCache_v3 import DataCache
|
||||
from ExchangeInterface import ExchangeInterface
|
||||
|
||||
# Object that interacts with the persistent data.
|
||||
data = DataCache()
|
||||
|
||||
# Object that interacts and maintains exchange_interface and account data
|
||||
exchanges = ExchangeInterface(data)
|
||||
|
||||
# Configuration and settings for the user app and charts
|
||||
config = Configuration()
|
||||
from Users import Users
|
||||
|
||||
|
||||
def test_get_indicators():
|
||||
# Todo test after creating indicator.
|
||||
result = config.users.get_indicators('guest_454')
|
||||
print('\n Test result:', result)
|
||||
assert type(result) is not None
|
||||
@pytest.fixture
|
||||
def data_cache():
|
||||
"""Create a fresh DataCache for each test."""
|
||||
return DataCache()
|
||||
|
||||
|
||||
def test_get_id():
|
||||
result = config.users.get_id('guest_454')
|
||||
print('\n Test result:', result)
|
||||
assert type(result) is int
|
||||
@pytest.fixture
|
||||
def users(data_cache):
|
||||
"""Create a Users instance with the data cache."""
|
||||
return Users(data_cache=data_cache)
|
||||
|
||||
|
||||
def test_get_username():
|
||||
result = config.users.get_username(28)
|
||||
print('\n Test result:', result)
|
||||
assert type(result) is str
|
||||
class TestUsers:
|
||||
"""Tests for the Users class."""
|
||||
|
||||
def test_create_guest(self, users):
|
||||
"""Test creating a guest user."""
|
||||
result = users.create_guest()
|
||||
print(f'\n Test result: {result}')
|
||||
assert isinstance(result, str)
|
||||
assert result.startswith('guest_')
|
||||
|
||||
def test_save_indicators():
|
||||
# ind=indicators.
|
||||
# result = config.users.save_indicators(indicator=ind)
|
||||
# print('\n Test result:', result)
|
||||
# assert type(result) is str
|
||||
pass
|
||||
def test_create_unique_guest_name(self, users):
|
||||
"""Test creating unique guest names."""
|
||||
result = users.create_unique_guest_name()
|
||||
print(f'\n Test result: {result}')
|
||||
assert isinstance(result, str)
|
||||
assert result.startswith('guest_')
|
||||
|
||||
def test_user_attr_is_taken(self, users):
|
||||
"""Test checking if user attribute is taken."""
|
||||
# Create a test user first
|
||||
users.create_guest()
|
||||
|
||||
def test_is_logged_in():
|
||||
session = {'user': 'guest_454'}
|
||||
result = config.users.is_logged_in(session.get('user'))
|
||||
print('\n Test result:', result)
|
||||
assert type(result) is bool
|
||||
# Check for a name that doesn't exist
|
||||
result = users.user_attr_is_taken('user_name', 'nonexistent_user_12345')
|
||||
print(f'\n Test result for nonexistent: {result}')
|
||||
assert isinstance(result, bool)
|
||||
assert result is False
|
||||
|
||||
def test_scramble_text(self, users):
|
||||
"""Test text scrambling (for password hashing)."""
|
||||
original = 'SomeText'
|
||||
result = users.scramble_text(original)
|
||||
print(f'\n Test result: {result}')
|
||||
assert isinstance(result, str)
|
||||
assert result != original # Should be different from original
|
||||
|
||||
def test_create_unique_guest_name():
|
||||
result = config.users.create_unique_guest_name()
|
||||
print('\n Test result:', result)
|
||||
assert type(result) is str
|
||||
|
||||
|
||||
def test_create_guest():
|
||||
result = config.users.create_guest()
|
||||
print('\n Test result:', result)
|
||||
assert type(result) is str
|
||||
|
||||
|
||||
def test_user_attr_is_taken():
|
||||
result = config.users.user_attr_is_taken('user_name', 'bill')
|
||||
print('\n Test result_1:', result)
|
||||
result = config.users.user_attr_is_taken('user_name', 'guest_374')
|
||||
print('\n Test result_2:', result)
|
||||
assert type(result) is bool
|
||||
|
||||
|
||||
def test_scramble_text():
|
||||
result = config.users.scramble_text('SomeText')
|
||||
print('\n Test result_2:', result)
|
||||
assert type(result) is str
|
||||
|
||||
|
||||
def test_create_new_user_in_db():
|
||||
result = config.users.create_new_user_in_db(({'user_name': 'Billy'},))
|
||||
print('\n Test result:', result)
|
||||
assert True
|
||||
|
||||
|
||||
def test_create_new_user():
|
||||
result = config.users.create_new_user(username='testy', email='yesy@email.com', password='hot_cow123')
|
||||
print('\n Test result:', result)
|
||||
def test_create_new_user(self, users):
|
||||
"""Test creating a new user with credentials."""
|
||||
import random
|
||||
username = f'test_user_{random.randint(1000, 9999)}'
|
||||
result = users.create_new_user(
|
||||
username=username,
|
||||
email=f'{username}@email.com',
|
||||
password='test_password123'
|
||||
)
|
||||
print(f'\n Test result: {result}')
|
||||
assert result is True
|
||||
|
||||
def test_load_or_create_user_creates_guest(self, users):
|
||||
"""Test that load_or_create_user creates a guest if user doesn't exist."""
|
||||
result = users.load_or_create_user(None)
|
||||
print(f'\n Test result: {result}')
|
||||
assert isinstance(result, str)
|
||||
assert result.startswith('guest_')
|
||||
|
||||
def test_load_or_create_user():
|
||||
session = {'user': 'guest_454'}
|
||||
result = config.users.load_or_create_user(session.get('user'))
|
||||
print('\n Test result:', result)
|
||||
assert type(result) is str
|
||||
|
||||
|
||||
def test_log_in_user():
|
||||
session = {'user': 'RobbieD'}
|
||||
result = config.users.log_in_user(session.get('user'), 'testPass1')
|
||||
print('\n Test result:', result)
|
||||
assert type(result) is bool
|
||||
|
||||
|
||||
def test_log_out_user():
|
||||
session = {'user': 'RobbieD'}
|
||||
result = config.users.log_user_in_out(session.get('user'))
|
||||
print('\n Test result:', result)
|
||||
assert type(result) is bool
|
||||
|
||||
|
||||
def test_log_out_all_users():
|
||||
result = config.users.log_out_all_users()
|
||||
print('\n Test result:', result)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_load_user_data():
|
||||
# Todo method incomplete
|
||||
result = config.users.get_user_data(user_name='RobbieD')
|
||||
print('\n Test result:', result)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_modify_user_data():
|
||||
# d = {"exchange": "alpaca", "timeframe": "5m", "market": "BTC/USD"}
|
||||
d=['alpaca']
|
||||
print(f'\n d:{d} of type: {type(d)}')
|
||||
my_data = json.dumps(d)
|
||||
result = config.users.modify_user_data(username='Billy', field_name='active_exchanges', new_data=my_data)
|
||||
print('\n Test result:', result)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_validate_password():
|
||||
result = config.users.validate_password(username='RobbieD', password='testPass1')
|
||||
print('\n Test result:', result)
|
||||
assert type(result) is bool
|
||||
|
||||
|
||||
def test_get_chart_view():
|
||||
result = config.users.get_chart_view(user_name='guest_454', prop='timeframe')
|
||||
print('\n Test result:', result)
|
||||
print('type:', type(result))
|
||||
|
||||
result = config.users.get_chart_view(user_name='guest_454', prop='exchange_name')
|
||||
print('\n Test result:', result)
|
||||
print('type:', type(result))
|
||||
|
||||
result = config.users.get_chart_view(user_name='guest_454', prop='market')
|
||||
print('\n Test result:', result)
|
||||
print('type:', type(result))
|
||||
def test_get_id_after_create(self, users):
|
||||
"""Test getting user ID after creation."""
|
||||
guest_name = users.create_guest()
|
||||
result = users.get_id(guest_name)
|
||||
print(f'\n Test result: {result}')
|
||||
assert result is not None
|
||||
assert isinstance(result, int)
|
||||
|
||||
result = config.users.get_chart_view(user_name='guest_454')
|
||||
print('\n Test result:', result)
|
||||
print('type:', type(result))
|
||||
assert result is not None
|
||||
def test_get_username_by_id(self, users):
|
||||
"""Test getting username by ID."""
|
||||
guest_name = users.create_guest()
|
||||
user_id = users.get_id(guest_name)
|
||||
result = users.get_username(user_id)
|
||||
print(f'\n Test result: {result}')
|
||||
assert result == guest_name
|
||||
|
||||
def test_is_logged_in_false(self, users):
|
||||
"""Test that a new guest is not logged in."""
|
||||
guest_name = users.create_guest()
|
||||
result = users.is_logged_in(guest_name)
|
||||
print(f'\n Test result: {result}')
|
||||
assert isinstance(result, bool)
|
||||
|
||||
def test_validate_password(self, users):
|
||||
"""Test password validation."""
|
||||
import random
|
||||
username = f'test_user_{random.randint(1000, 9999)}'
|
||||
password = 'correct_password123'
|
||||
users.create_new_user(username=username, email=f'{username}@test.com', password=password)
|
||||
|
||||
# Should return True for correct password
|
||||
result = users.validate_password(username=username, password=password)
|
||||
print(f'\n Test result correct password: {result}')
|
||||
assert result is True
|
||||
|
||||
# Should return False for wrong password
|
||||
result = users.validate_password(username=username, password='wrong_password')
|
||||
print(f'\n Test result wrong password: {result}')
|
||||
assert result is False
|
||||
|
||||
def test_get_chart_view(self, users):
|
||||
"""Test getting chart view settings."""
|
||||
guest_name = users.create_guest()
|
||||
|
||||
# Get specific property
|
||||
result = users.get_chart_view(user_name=guest_name, prop='timeframe')
|
||||
print(f'\n Test result timeframe: {result}')
|
||||
|
||||
# Get all chart view settings
|
||||
result = users.get_chart_view(user_name=guest_name)
|
||||
print(f'\n Test result all: {result}')
|
||||
# Result can be None or a dict, both are valid
|
||||
|
||||
def test_log_out_all_users(self, users):
|
||||
"""Test logging out all users."""
|
||||
result = users.log_out_all_users()
|
||||
print(f'\n Test result: {result}')
|
||||
# Should complete without error
|
||||
|
||||
def test_save_and_get_indicators(self, users, data_cache):
|
||||
"""Test saving and retrieving indicators."""
|
||||
# Create the indicators cache if it doesn't exist
|
||||
if 'indicators' not in data_cache.caches:
|
||||
data_cache.create_cache('indicators', cache_type='row')
|
||||
|
||||
guest_name = users.create_guest()
|
||||
|
||||
# Get indicators (may be empty initially)
|
||||
result = users.get_indicators(guest_name)
|
||||
print(f'\n Test result get: {result}')
|
||||
# Result can be None or a DataFrame, both are valid
|
||||
|
|
|
|||
|
|
@ -20,26 +20,36 @@ class FlaskAppTests(unittest.TestCase):
|
|||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn(b'Welcome', response.data) # Adjust this based on your actual landing page content
|
||||
|
||||
def test_login(self):
|
||||
def test_login_redirect_on_invalid(self):
|
||||
"""
|
||||
Test the login route with valid and invalid credentials.
|
||||
Test that login redirects on invalid credentials.
|
||||
"""
|
||||
# Valid credentials
|
||||
valid_data = {'user_name': 'test_user', 'password': 'test_password'}
|
||||
response = self.app.post('/login', data=valid_data)
|
||||
self.assertEqual(response.status_code, 302) # Redirects on success
|
||||
|
||||
# Invalid credentials
|
||||
# Invalid credentials should redirect to login page
|
||||
invalid_data = {'user_name': 'wrong_user', 'password': 'wrong_password'}
|
||||
response = self.app.post('/login', data=invalid_data)
|
||||
self.assertEqual(response.status_code, 302) # Redirects on failure
|
||||
self.assertIn(b'Invalid user_name or password', response.data)
|
||||
# Follow redirect to check error flash message
|
||||
response = self.app.post('/login', data=invalid_data, follow_redirects=True)
|
||||
# The page should contain login form or error message
|
||||
self.assertIn(b'login', response.data.lower())
|
||||
|
||||
def test_login_with_valid_credentials(self):
|
||||
"""
|
||||
Test the login route with credentials that may or may not exist.
|
||||
"""
|
||||
# Valid credentials (test user may not exist)
|
||||
valid_data = {'user_name': 'test_user', 'password': 'test_password'}
|
||||
response = self.app.post('/login', data=valid_data)
|
||||
# Should redirect regardless (to index if success, to login if failure)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
|
||||
def test_signup(self):
|
||||
"""
|
||||
Test the signup route.
|
||||
"""
|
||||
data = {'email': 'test@example.com', 'user_name': 'new_user', 'password': 'new_password'}
|
||||
import random
|
||||
username = f'test_user_{random.randint(10000, 99999)}'
|
||||
data = {'email': f'{username}@example.com', 'user_name': username, 'password': 'new_password'}
|
||||
response = self.app.post('/signup_submit', data=data)
|
||||
self.assertEqual(response.status_code, 302) # Redirects on success
|
||||
|
||||
|
|
@ -50,23 +60,28 @@ class FlaskAppTests(unittest.TestCase):
|
|||
response = self.app.get('/signout')
|
||||
self.assertEqual(response.status_code, 302) # Redirects on signout
|
||||
|
||||
def test_history(self):
|
||||
def test_indicator_init_requires_auth(self):
|
||||
"""
|
||||
Test the history route.
|
||||
"""
|
||||
data = {"user_name": "test_user"}
|
||||
response = self.app.post('/api/history', data=json.dumps(data), content_type='application/json')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn(b'price_history', response.data)
|
||||
|
||||
def test_indicator_init(self):
|
||||
"""
|
||||
Test the indicator initialization route.
|
||||
Test that indicator_init requires authentication.
|
||||
"""
|
||||
data = {"user_name": "test_user"}
|
||||
response = self.app.post('/api/indicator_init', data=json.dumps(data), content_type='application/json')
|
||||
# Should return 401 without proper session
|
||||
self.assertIn(response.status_code, [200, 401]) # Either authenticated or not
|
||||
|
||||
def test_login_page_loads(self):
|
||||
"""
|
||||
Test that login page loads.
|
||||
"""
|
||||
response = self.app.get('/login')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_signup_page_loads(self):
|
||||
"""
|
||||
Test that signup page loads.
|
||||
"""
|
||||
response = self.app.get('/signup')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn(b'indicator_data', response.data)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -1,195 +1,177 @@
|
|||
import datetime
|
||||
"""
|
||||
Tests for the Candles module.
|
||||
|
||||
Note: The candle data fetching has been moved to EDM (Exchange Data Manager).
|
||||
These tests use mocked EDM responses to avoid requiring a running EDM server.
|
||||
Functions like ts_of_n_minutes_ago and timeframe_to_minutes are tested in
|
||||
test_shared_utilities.py.
|
||||
"""
|
||||
import datetime as dt
|
||||
import pytest
|
||||
import pandas as pd
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from candles import Candles
|
||||
from Configuration import Configuration
|
||||
from ExchangeInterface import ExchangeInterface
|
||||
from DataCache_v3 import DataCache
|
||||
|
||||
|
||||
def test_sqlite():
|
||||
# Object that interacts and maintains exchange_interface and account data
|
||||
exchanges = ExchangeInterface()
|
||||
# Configuration and settings for the user app and charts
|
||||
conf = Configuration()
|
||||
candle_obj = Candles(exchanges=exchanges, config_obj=conf)
|
||||
assert candle_obj is not None
|
||||
# candle_obj.set_candle_history(symbol='BTCUSDT',
|
||||
# timeframe='15m',
|
||||
# exchange_name='binance_futures')
|
||||
@pytest.fixture
|
||||
def mock_edm_client():
|
||||
"""Create a mock EDM client."""
|
||||
mock = MagicMock()
|
||||
# Mock get_candles_sync to return sample candle data
|
||||
sample_candles = pd.DataFrame({
|
||||
'time': [1700000000000, 1700000060000, 1700000120000, 1700000180000, 1700000240000],
|
||||
'open': [100.0, 101.0, 102.0, 101.5, 103.0],
|
||||
'high': [102.0, 103.0, 104.0, 103.5, 105.0],
|
||||
'low': [99.0, 100.0, 101.0, 100.5, 102.0],
|
||||
'close': [101.0, 102.0, 101.5, 103.0, 104.0],
|
||||
'volume': [1000.0, 1200.0, 800.0, 1500.0, 900.0]
|
||||
})
|
||||
mock.get_candles_sync.return_value = sample_candles
|
||||
return mock
|
||||
|
||||
|
||||
def test_ts_of_n_minutes_ago():
|
||||
ca = 12
|
||||
lc = 5
|
||||
print(f'\ncandles ago is {ca}, length of candle is {lc}:')
|
||||
result = ts_of_n_minutes_ago(ca, lc)
|
||||
print(f'Result: TYPE({type(result)}): {result}')
|
||||
print(f'The time right now is: {datetime.datetime.now()}')
|
||||
difference = datetime.datetime.now() - result
|
||||
print(f'The difference is: {difference}')
|
||||
# Result is supposed to be one candle length extra.
|
||||
assert difference == datetime.timedelta(hours=1, minutes=5, seconds=00)
|
||||
@pytest.fixture
|
||||
def mock_exchanges():
|
||||
"""Create mock exchange interface."""
|
||||
return MagicMock()
|
||||
|
||||
|
||||
def test_timeframe_to_minutes():
|
||||
result = timeframe_to_minutes('1w')
|
||||
print(f'\nnumber of minutes: {result}')
|
||||
assert result == 10080
|
||||
@pytest.fixture
|
||||
def mock_users():
|
||||
"""Create mock users object."""
|
||||
mock = MagicMock()
|
||||
mock.get_chart_view.return_value = {'timeframe': '1m', 'exchange': 'binance', 'market': 'BTC/USDT'}
|
||||
return mock
|
||||
|
||||
|
||||
def test_fetch_exchange_id():
|
||||
exchange_id = fetch_exchange_id('binance_coin')
|
||||
print(f'\nexchange_id: {exchange_id}')
|
||||
assert exchange_id == 3
|
||||
@pytest.fixture
|
||||
def data_cache():
|
||||
"""Create a fresh DataCache."""
|
||||
return DataCache()
|
||||
|
||||
|
||||
def test_fetch_market_id_from_db():
|
||||
market_id = fetch_market_id_from_db('ETHUSDT', 'binance_spot')
|
||||
print(f'\nmarket_id: {market_id}')
|
||||
assert market_id == 1
|
||||
@pytest.fixture
|
||||
def config():
|
||||
"""Create a Configuration instance."""
|
||||
return Configuration()
|
||||
|
||||
|
||||
def test_fetch_candles_from_exchange():
|
||||
# Object that interacts and maintains exchange_interface and account data
|
||||
exchanges = ExchangeInterface()
|
||||
candles_obj = Candles(exchanges)
|
||||
|
||||
result = candles_obj._fetch_candles_from_exchange(symbol='BTCUSDT', interval='15m', exchange_name='binance_spot',
|
||||
start_datetime={'year': 2023, 'month': 3, 'day': 15})
|
||||
|
||||
print(f'\n{result.head().to_string()}')
|
||||
|
||||
result = candles_obj._fetch_candles_from_exchange(symbol='BTCUSDT', interval='15m', exchange_name='binance_futures',
|
||||
start_datetime={'year': 2023, 'month': 3, 'day': 15})
|
||||
print(f'\n{result.head().to_string()}')
|
||||
|
||||
result = candles_obj._fetch_candles_from_exchange(symbol='BTCUSD_PERP', interval='15m',
|
||||
exchange_name='binance_coin',
|
||||
start_datetime={'year': 2023, 'month': 3, 'day': 15})
|
||||
print(f'\n{result.head().to_string()}')
|
||||
|
||||
result = candles_obj._fetch_candles_from_exchange(symbol='BTC/USDT', interval='15m', exchange_name='alpaca',
|
||||
start_datetime={'year': 2023, 'month': 3, 'day': 15})
|
||||
print(f'\n{result.head().to_string()}')
|
||||
assert not result.empty
|
||||
@pytest.fixture
|
||||
def candles(mock_exchanges, mock_users, data_cache, config, mock_edm_client):
|
||||
"""Create a Candles instance with mocked dependencies."""
|
||||
return Candles(
|
||||
exchanges=mock_exchanges,
|
||||
users=mock_users,
|
||||
datacache=data_cache,
|
||||
config=config,
|
||||
edm_client=mock_edm_client
|
||||
)
|
||||
|
||||
|
||||
def test_insert_candles_into_db():
|
||||
# Object that interacts and maintains exchange_interface and account data
|
||||
exchanges = ExchangeInterface()
|
||||
candles_obj = Candles(exchanges)
|
||||
symbol = 'ETHUSDT'
|
||||
interval = '15m'
|
||||
exchange_name = 'binance_spot'
|
||||
result = candles_obj._fetch_candles_from_exchange(symbol=symbol, interval=interval, exchange_name=exchange_name,
|
||||
start_datetime={'year': 2023, 'month': 3, 'day': 15})
|
||||
class TestCandles:
|
||||
"""Tests for the Candles class."""
|
||||
|
||||
insert_candles_into_db(result, symbol=symbol, interval=interval, exchange_name=exchange_name)
|
||||
assert True
|
||||
def test_candles_creation(self, candles):
|
||||
"""Test that Candles object can be created."""
|
||||
assert candles is not None
|
||||
|
||||
|
||||
def test_get_db_records_since():
|
||||
timestamp = datetime.datetime.timestamp(datetime.datetime(year=2023, month=3, day=14, hour=1, minute=0)) * 1000
|
||||
result = get_db_records_since(table_name='ETHUSDT_15m_alpaca', timestamp=timestamp)
|
||||
print(result.head())
|
||||
assert result is not None
|
||||
|
||||
|
||||
def test_get_from_cache_since():
|
||||
# Object that interacts and maintains exchange_interface and account data
|
||||
exchanges = ExchangeInterface()
|
||||
candles_obj = Candles(exchanges)
|
||||
symbol = 'ETHUSDT'
|
||||
interval = '15m'
|
||||
exchange_name = 'binance_spot'
|
||||
timestamp = datetime.datetime.timestamp(datetime.datetime(year=2023, month=3, day=16, hour=1, minute=0)) * 1000
|
||||
key = f'{symbol}_{interval}_{exchange_name}'
|
||||
|
||||
result = candles_obj.get_from_cache_since(key=key, start_datetime=timestamp)
|
||||
print(result)
|
||||
assert result is not None
|
||||
|
||||
|
||||
def test_get_records_since():
|
||||
# Object that interacts and maintains exchange_interface and account data
|
||||
exchanges = ExchangeInterface()
|
||||
candles_obj = Candles(exchanges)
|
||||
symbol = 'BTCUSDT'
|
||||
interval = '2h'
|
||||
exchange_name = 'binance_spot'
|
||||
|
||||
start_time = datetime.datetime(year=2023, month=3, day=14, hour=1, minute=0)
|
||||
print(f'\ntest_candles_get_records() starting @: {start_time}')
|
||||
result = candles_obj.get_records_since(symbol=symbol, timeframe=interval,
|
||||
exchange_name=exchange_name, start_time=start_time)
|
||||
print(f'test_candles_received records from: {datetime.datetime.fromtimestamp(result.time.min() / 1000)}')
|
||||
|
||||
start_time = datetime.datetime(year=2023, month=3, day=16, hour=1, minute=0)
|
||||
print(f'\ntest_candles_get_records() starting @: {start_time}')
|
||||
result = candles_obj.get_records_since(symbol=symbol, timeframe=interval,
|
||||
exchange_name=exchange_name, start_time=start_time)
|
||||
print(f'test_candles_received records from: {datetime.datetime.fromtimestamp(result.time.min() / 1000)}')
|
||||
|
||||
start_time = datetime.datetime(year=2023, month=3, day=13, hour=1, minute=0)
|
||||
print(f'\ntest_candles_get_records() starting @: {start_time}')
|
||||
result = candles_obj.get_records_since(symbol=symbol, timeframe=interval,
|
||||
exchange_name=exchange_name, start_time=start_time)
|
||||
print(f'test_candles_received records from: {datetime.datetime.fromtimestamp(result.time.min() / 1000)}')
|
||||
def test_get_last_n_candles(self, candles, mock_edm_client):
|
||||
"""Test getting last N candles."""
|
||||
result = candles.get_last_n_candles(
|
||||
num_candles=5,
|
||||
asset='BTC/USDT',
|
||||
timeframe='1m',
|
||||
exchange='binance',
|
||||
user_name='test_user'
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
assert isinstance(result, pd.DataFrame)
|
||||
assert len(result) == 5
|
||||
assert 'open' in result.columns
|
||||
assert 'high' in result.columns
|
||||
assert 'low' in result.columns
|
||||
assert 'close' in result.columns
|
||||
assert 'volume' in result.columns
|
||||
|
||||
def test_get_last_n_candles_caps_at_1000(self, candles, mock_edm_client):
|
||||
"""Test that requests for more than 1000 candles are capped."""
|
||||
candles.get_last_n_candles(
|
||||
num_candles=2000,
|
||||
asset='BTC/USDT',
|
||||
timeframe='1m',
|
||||
exchange='binance',
|
||||
user_name='test_user'
|
||||
)
|
||||
|
||||
# Check that EDM was called with capped limit
|
||||
call_args = mock_edm_client.get_candles_sync.call_args
|
||||
assert call_args.kwargs.get('limit', call_args[1].get('limit')) == 1000
|
||||
|
||||
def test_candles_raises_without_edm(self, mock_exchanges, mock_users, data_cache, config):
|
||||
"""Test that Candles raises error when EDM is not available."""
|
||||
candles_no_edm = Candles(
|
||||
exchanges=mock_exchanges,
|
||||
users=mock_users,
|
||||
datacache=data_cache,
|
||||
config=config,
|
||||
edm_client=None
|
||||
)
|
||||
|
||||
with pytest.raises(RuntimeError, match="EDM client not initialized"):
|
||||
candles_no_edm.get_last_n_candles(
|
||||
num_candles=5,
|
||||
asset='BTC/USDT',
|
||||
timeframe='1m',
|
||||
exchange='binance',
|
||||
user_name='test_user'
|
||||
)
|
||||
|
||||
def test_get_latest_values(self, candles, mock_edm_client):
|
||||
"""Test getting latest values for a specific field."""
|
||||
result = candles.get_latest_values(
|
||||
value_name='close',
|
||||
symbol='BTC/USDT',
|
||||
timeframe='1m',
|
||||
exchange='binance',
|
||||
num_record=5,
|
||||
user_name='test_user'
|
||||
)
|
||||
|
||||
def test_get_last_n_candles():
|
||||
# Object that interacts and maintains exchange_interface and account data
|
||||
exchanges = ExchangeInterface()
|
||||
candles_obj = Candles(exchanges)
|
||||
symbol = 'ETHUSDT'
|
||||
interval = '1m'
|
||||
exchange_name = 'binance_spot'
|
||||
print(f'\n Here we go!')
|
||||
result = candles_obj.get_last_n_candles(5, symbol, interval, exchange_name)
|
||||
print(result)
|
||||
result = candles_obj.get_last_n_candles(10, symbol, interval, exchange_name)
|
||||
print(result)
|
||||
result = candles_obj.get_last_n_candles(40, symbol, interval, exchange_name)
|
||||
print(result)
|
||||
result = candles_obj.get_last_n_candles(20, symbol, interval, exchange_name)
|
||||
print(result)
|
||||
result = candles_obj.get_last_n_candles(15, symbol, interval, exchange_name)
|
||||
print(result)
|
||||
assert result is not None
|
||||
# Should return series/list of close values
|
||||
assert len(result) <= 5
|
||||
|
||||
def test_max_records_from_config(self, candles, config):
|
||||
"""Test that max_records is loaded from config."""
|
||||
expected_max = config.get_setting('max_data_loaded')
|
||||
assert candles.max_records == expected_max
|
||||
|
||||
def test_candle_cache_created(self, candles, data_cache):
|
||||
"""Test that candle cache is created on initialization."""
|
||||
# The cache should exist
|
||||
assert 'candles' in data_cache.caches
|
||||
|
||||
|
||||
def test_get_latest_values():
|
||||
# Object that interacts and maintains exchange_interface and account data
|
||||
exchanges = ExchangeInterface()
|
||||
# Configuration and settings for the user app and charts
|
||||
conf = Configuration()
|
||||
candle_obj = Candles(exchanges=exchanges, config_obj=conf)
|
||||
class TestCandlesIntegration:
|
||||
"""Integration-style tests that verify EDM client interaction."""
|
||||
|
||||
symbol = 'ETHUSDT'
|
||||
interval = '1m'
|
||||
exchange_name = 'binance_spot'
|
||||
result = candle_obj.get_latest_values('open', symbol=symbol, timeframe=interval,
|
||||
exchange=exchange_name, num_record=10)
|
||||
print(result)
|
||||
assert result is not None
|
||||
def test_edm_client_called_correctly(self, candles, mock_edm_client):
|
||||
"""Test that EDM client is called with correct parameters."""
|
||||
candles.get_last_n_candles(
|
||||
num_candles=10,
|
||||
asset='ETH/USDT',
|
||||
timeframe='5m',
|
||||
exchange='binance',
|
||||
user_name='test_user'
|
||||
)
|
||||
|
||||
|
||||
def test_get_colour_coded_volume():
|
||||
# Object that interacts and maintains exchange_interface and account data
|
||||
exchanges = ExchangeInterface()
|
||||
# Configuration and settings for the user app and charts
|
||||
conf = Configuration()
|
||||
candle_obj = Candles(exchanges=exchanges, config_obj=conf)
|
||||
|
||||
symbol = 'ETHUSDT'
|
||||
interval = '1h'
|
||||
exchange_name = 'binance_spot'
|
||||
result = candle_obj.get_latest_values('volume', symbol=symbol, timeframe=interval,
|
||||
exchange=exchange_name, num_record=10)
|
||||
print(result)
|
||||
assert result is not None
|
||||
|
||||
|
||||
def test_get_last_n_candles():
|
||||
assert False
|
||||
mock_edm_client.get_candles_sync.assert_called_once()
|
||||
call_kwargs = mock_edm_client.get_candles_sync.call_args.kwargs
|
||||
assert call_kwargs['symbol'] == 'ETH/USDT'
|
||||
assert call_kwargs['timeframe'] == '5m'
|
||||
assert call_kwargs['exchange'] == 'binance'
|
||||
assert call_kwargs['limit'] == 10
|
||||
|
|
|
|||
|
|
@ -6,6 +6,11 @@ from Database import Database, SQLite, make_query, make_insert, HDict
|
|||
from shared_utilities import unix_time_millis
|
||||
|
||||
|
||||
def utcnow() -> dt.datetime:
|
||||
"""Return timezone-aware UTC datetime."""
|
||||
return dt.datetime.now(dt.timezone.utc)
|
||||
|
||||
|
||||
class TestSQLite(unittest.TestCase):
|
||||
def test_sqlite_context_manager(self):
|
||||
print("\nRunning test_sqlite_context_manager...")
|
||||
|
|
@ -56,7 +61,7 @@ class TestDatabase(unittest.TestCase):
|
|||
def test_make_insert(self):
|
||||
print("\nRunning test_make_insert...")
|
||||
insert = make_insert('test_table', ('name', 'age'))
|
||||
expected_insert = "INSERT INTO test_table ('name', 'age') VALUES(?, ?);"
|
||||
expected_insert = 'INSERT INTO "test_table" ("name", "age") VALUES (?, ?);'
|
||||
self.assertEqual(insert, expected_insert)
|
||||
print("Make insert test passed.")
|
||||
|
||||
|
|
@ -74,7 +79,7 @@ class TestDatabase(unittest.TestCase):
|
|||
self.cursor.execute('CREATE TABLE test_table (id INTEGER PRIMARY KEY, name TEXT)')
|
||||
self.cursor.execute("INSERT INTO test_table (id, name) VALUES (1, 'test')")
|
||||
self.connection.commit()
|
||||
rows = self.db.get_rows_where('test_table', ('name', 'test'))
|
||||
rows = self.db.get_rows_where('test_table', [('name', 'test')])
|
||||
self.assertIsInstance(rows, pd.DataFrame)
|
||||
self.assertEqual(rows.iloc[0]['name'], 'test')
|
||||
print("Get rows where test passed.")
|
||||
|
|
@ -111,7 +116,7 @@ class TestDatabase(unittest.TestCase):
|
|||
def test_get_timestamped_records(self):
|
||||
print("\nRunning test_get_timestamped_records...")
|
||||
df = pd.DataFrame({
|
||||
'time': [unix_time_millis(dt.datetime.utcnow())],
|
||||
'time': [unix_time_millis(utcnow())],
|
||||
'open': [1.0],
|
||||
'high': [1.0],
|
||||
'low': [1.0],
|
||||
|
|
@ -132,8 +137,8 @@ class TestDatabase(unittest.TestCase):
|
|||
""")
|
||||
self.connection.commit()
|
||||
self.db.insert_dataframe(df, table_name)
|
||||
st = dt.datetime.utcnow() - dt.timedelta(minutes=1)
|
||||
et = dt.datetime.utcnow()
|
||||
st = utcnow() - dt.timedelta(minutes=1)
|
||||
et = utcnow()
|
||||
records = self.db.get_timestamped_records(table_name, 'time', st, et)
|
||||
self.assertIsInstance(records, pd.DataFrame)
|
||||
self.assertFalse(records.empty)
|
||||
|
|
@ -153,7 +158,7 @@ class TestDatabase(unittest.TestCase):
|
|||
def test_insert_candles_into_db(self):
|
||||
print("\nRunning test_insert_candles_into_db...")
|
||||
df = pd.DataFrame({
|
||||
'time': [unix_time_millis(dt.datetime.utcnow())],
|
||||
'time': [unix_time_millis(utcnow())],
|
||||
'open': [1.0],
|
||||
'high': [1.0],
|
||||
'low': [1.0],
|
||||
|
|
|
|||
|
|
@ -1,9 +1,10 @@
|
|||
import logging
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from unittest.mock import MagicMock, patch, PropertyMock
|
||||
from datetime import datetime
|
||||
from ExchangeInterface import ExchangeInterface
|
||||
from Exchange import Exchange
|
||||
from DataCache_v3 import DataCache
|
||||
from typing import Dict, Any
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
|
||||
|
|
@ -23,22 +24,15 @@ class Trade:
|
|||
|
||||
class TestExchangeInterface(unittest.TestCase):
|
||||
|
||||
@patch('Exchange.Exchange')
|
||||
def setUp(self, MockExchange):
|
||||
|
||||
self.exchange_interface = ExchangeInterface()
|
||||
|
||||
# Mock exchange instances
|
||||
self.mock_exchange = MockExchange.return_value
|
||||
def setUp(self):
|
||||
self.cache_manager = DataCache()
|
||||
self.exchange_interface = ExchangeInterface(self.cache_manager)
|
||||
|
||||
# Setup test data
|
||||
self.user_name = "test_user"
|
||||
self.exchange_name = "binance"
|
||||
self.api_keys = {'key': 'test_key', 'secret': 'test_secret'}
|
||||
|
||||
# Connect the mock exchange
|
||||
self.exchange_interface.connect_exchange(self.exchange_name, self.user_name, self.api_keys)
|
||||
|
||||
# Mock trade object
|
||||
self.trade = Trade(target=self.exchange_name, symbol="BTC/USDT", order_id="12345")
|
||||
|
||||
|
|
@ -50,69 +44,70 @@ class TestExchangeInterface(unittest.TestCase):
|
|||
}
|
||||
|
||||
def test_get_trade_status(self):
|
||||
self.mock_exchange.get_order.return_value = self.order_data
|
||||
assert isinstance(self.mock_exchange.get_order.return_value, dict) # Ensure return value is dict
|
||||
"""Test getting trade status with mocked exchange."""
|
||||
mock_exchange = MagicMock()
|
||||
mock_exchange.get_order.return_value = self.order_data
|
||||
|
||||
with self.assertLogs(level='ERROR') as log:
|
||||
# Mock get_exchange to return our mock
|
||||
with patch.object(self.exchange_interface, 'get_exchange', return_value=mock_exchange):
|
||||
status = self.exchange_interface.get_trade_info(self.trade, self.user_name, 'status')
|
||||
if any('Must configure API keys' in message for message in log.output):
|
||||
return
|
||||
self.assertEqual(status, 'closed')
|
||||
|
||||
def test_get_trade_executed_qty(self):
|
||||
self.mock_exchange.get_order.return_value = self.order_data
|
||||
assert isinstance(self.mock_exchange.get_order.return_value, dict) # Ensure return value is dict
|
||||
"""Test getting executed quantity with mocked exchange."""
|
||||
mock_exchange = MagicMock()
|
||||
mock_exchange.get_order.return_value = self.order_data
|
||||
|
||||
with self.assertLogs(level='ERROR') as log:
|
||||
with patch.object(self.exchange_interface, 'get_exchange', return_value=mock_exchange):
|
||||
executed_qty = self.exchange_interface.get_trade_info(self.trade, self.user_name, 'executed_qty')
|
||||
if any('Must configure API keys' in message for message in log.output):
|
||||
return
|
||||
self.assertEqual(executed_qty, 1.0)
|
||||
|
||||
def test_get_trade_executed_price(self):
|
||||
self.mock_exchange.get_order.return_value = self.order_data
|
||||
assert isinstance(self.mock_exchange.get_order.return_value, dict) # Ensure return value is dict
|
||||
"""Test getting executed price with mocked exchange."""
|
||||
mock_exchange = MagicMock()
|
||||
mock_exchange.get_order.return_value = self.order_data
|
||||
|
||||
with self.assertLogs(level='ERROR') as log:
|
||||
with patch.object(self.exchange_interface, 'get_exchange', return_value=mock_exchange):
|
||||
executed_price = self.exchange_interface.get_trade_info(self.trade, self.user_name, 'executed_price')
|
||||
if any('Must configure API keys' in message for message in log.output):
|
||||
return
|
||||
self.assertEqual(executed_price, 50000.0)
|
||||
|
||||
def test_invalid_info_type(self):
|
||||
self.mock_exchange.get_order.return_value = self.order_data
|
||||
assert isinstance(self.mock_exchange.get_order.return_value, dict) # Ensure return value is dict
|
||||
"""Test invalid info type returns None."""
|
||||
mock_exchange = MagicMock()
|
||||
mock_exchange.get_order.return_value = self.order_data
|
||||
|
||||
with patch.object(self.exchange_interface, 'get_exchange', return_value=mock_exchange):
|
||||
with self.assertLogs(level='ERROR') as log:
|
||||
result = self.exchange_interface.get_trade_info(self.trade, self.user_name, 'invalid_type')
|
||||
if any('Must configure API keys' in message for message in log.output):
|
||||
return
|
||||
self.assertIsNone(result)
|
||||
self.assertTrue(any('Invalid info type' in message for message in log.output))
|
||||
|
||||
def test_order_not_found(self):
|
||||
self.mock_exchange.get_order.return_value = None
|
||||
"""Test order not found returns None."""
|
||||
mock_exchange = MagicMock()
|
||||
mock_exchange.get_order.return_value = None
|
||||
|
||||
with patch.object(self.exchange_interface, 'get_exchange', return_value=mock_exchange):
|
||||
with self.assertLogs(level='ERROR') as log:
|
||||
result = self.exchange_interface.get_trade_info(self.trade, self.user_name, 'status')
|
||||
if any('Must configure API keys' in message for message in log.output):
|
||||
return
|
||||
self.assertIsNone(result)
|
||||
self.assertTrue(any('Order 12345 for BTC/USDT not found.' in message for message in log.output))
|
||||
self.assertTrue(any('not found' in message for message in log.output))
|
||||
|
||||
def test_get_price_default_source(self):
|
||||
# Setup the mock to return a specific price
|
||||
symbol = "BTC/USD"
|
||||
price = self.exchange_interface.get_price(symbol)
|
||||
"""Test get_price with default exchange."""
|
||||
mock_exchange = MagicMock()
|
||||
mock_exchange.get_price.return_value = 50000.0
|
||||
|
||||
self.assertLess(0.1, price)
|
||||
with patch.object(self.exchange_interface, 'connect_default_exchange'):
|
||||
self.exchange_interface.default_exchange = mock_exchange
|
||||
price = self.exchange_interface.get_price("BTC/USDT")
|
||||
self.assertEqual(price, 50000.0)
|
||||
|
||||
def test_get_price_with_invalid_source(self):
|
||||
symbol = "BTC/USD"
|
||||
with self.assertRaises(ValueError) as context:
|
||||
self.exchange_interface.get_price(symbol, price_source="invalid_source")
|
||||
|
||||
self.assertTrue('No implementation for price source: invalid_source' in str(context.exception))
|
||||
def test_get_price_with_invalid_exchange(self):
|
||||
"""Test get_price with invalid exchange name returns 0."""
|
||||
# Unknown exchange should return 0.0
|
||||
price = self.exchange_interface.get_price("BTC/USDT", exchange_name="invalid_exchange_xyz")
|
||||
self.assertEqual(price, 0.0)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -2,11 +2,17 @@ import unittest
|
|||
import ccxt
|
||||
from datetime import datetime
|
||||
import pandas as pd
|
||||
import pytest
|
||||
from Exchange import Exchange
|
||||
from typing import Dict, Optional
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestExchange(unittest.TestCase):
|
||||
"""
|
||||
Integration tests that connect to real exchanges.
|
||||
Run with: pytest -m integration tests/test_live_exchange_integration.py
|
||||
"""
|
||||
api_keys: Optional[Dict[str, str]]
|
||||
exchange: Exchange
|
||||
|
||||
|
|
|
|||
|
|
@ -8,6 +8,11 @@ from shared_utilities import (
|
|||
)
|
||||
|
||||
|
||||
def utcnow() -> dt.datetime:
|
||||
"""Return timezone-aware UTC datetime."""
|
||||
return dt.datetime.now(dt.timezone.utc)
|
||||
|
||||
|
||||
class TestSharedUtilities(unittest.TestCase):
|
||||
|
||||
def test_query_uptodate(self):
|
||||
|
|
@ -26,7 +31,7 @@ class TestSharedUtilities(unittest.TestCase):
|
|||
self.assertIsNotNone(result)
|
||||
|
||||
# (Test case 2) The records should be up-to-date (recent timestamps)
|
||||
now = unix_time_millis(dt.datetime.utcnow())
|
||||
now = unix_time_millis(utcnow())
|
||||
recent_records = pd.DataFrame({
|
||||
'time': [now - 70000, now - 60000, now - 40000]
|
||||
})
|
||||
|
|
@ -42,7 +47,7 @@ class TestSharedUtilities(unittest.TestCase):
|
|||
# The records should not be up-to-date (recent timestamps)
|
||||
one_hour = 60 * 60 * 1000 # one hour in milliseconds
|
||||
tolerance_milliseconds = 10 * 1000 # tolerance in milliseconds
|
||||
recent_time = unix_time_millis(dt.datetime.utcnow())
|
||||
recent_time = unix_time_millis(utcnow())
|
||||
borderline_records = pd.DataFrame({
|
||||
'time': [recent_time - one_hour + (tolerance_milliseconds - 3)] # just within the tolerance
|
||||
})
|
||||
|
|
@ -58,7 +63,7 @@ class TestSharedUtilities(unittest.TestCase):
|
|||
# The records should be up-to-date (recent timestamps)
|
||||
one_hour = 60 * 60 * 1000 # one hour in milliseconds
|
||||
tolerance_milliseconds = 10 * 1000 # tolerance in milliseconds
|
||||
recent_time = unix_time_millis(dt.datetime.utcnow())
|
||||
recent_time = unix_time_millis(utcnow())
|
||||
borderline_records = pd.DataFrame({
|
||||
'time': [recent_time - one_hour + (tolerance_milliseconds + 3)] # just within the tolerance
|
||||
})
|
||||
|
|
@ -77,19 +82,19 @@ class TestSharedUtilities(unittest.TestCase):
|
|||
|
||||
def test_unix_time_seconds(self):
|
||||
print('Testing unix_time_seconds()')
|
||||
time = dt.datetime(2020, 1, 1)
|
||||
time = dt.datetime(2020, 1, 1, tzinfo=dt.timezone.utc)
|
||||
self.assertEqual(unix_time_seconds(time), 1577836800)
|
||||
|
||||
def test_unix_time_millis(self):
|
||||
print('Testing unix_time_millis()')
|
||||
time = dt.datetime(2020, 1, 1)
|
||||
time = dt.datetime(2020, 1, 1, tzinfo=dt.timezone.utc)
|
||||
self.assertEqual(unix_time_millis(time), 1577836800000.0)
|
||||
|
||||
def test_query_satisfied(self):
|
||||
print('Testing query_satisfied()')
|
||||
|
||||
# Test case where the records should satisfy the query (records cover the start time)
|
||||
start_datetime = dt.datetime(2020, 1, 1)
|
||||
start_datetime = dt.datetime(2020, 1, 1, tzinfo=dt.timezone.utc)
|
||||
records = pd.DataFrame({
|
||||
'time': [1577836800000.0 - 600000, 1577836800000.0 - 300000, 1577836800000.0]
|
||||
# Covering the start time
|
||||
|
|
@ -103,7 +108,7 @@ class TestSharedUtilities(unittest.TestCase):
|
|||
self.assertIsNotNone(result)
|
||||
|
||||
# Test case where the records should not satisfy the query (recent records but not enough)
|
||||
recent_time = unix_time_millis(dt.datetime.utcnow())
|
||||
recent_time = unix_time_millis(utcnow())
|
||||
records = pd.DataFrame({
|
||||
'time': [recent_time - 300 * 60 * 1000, recent_time - 240 * 60 * 1000, recent_time - 180 * 60 * 1000]
|
||||
})
|
||||
|
|
@ -116,11 +121,11 @@ class TestSharedUtilities(unittest.TestCase):
|
|||
self.assertIsNone(result)
|
||||
|
||||
# Additional test case for partial coverage
|
||||
start_datetime = dt.datetime.utcnow() - dt.timedelta(minutes=300)
|
||||
start_datetime = utcnow() - dt.timedelta(minutes=300)
|
||||
records = pd.DataFrame({
|
||||
'time': [unix_time_millis(dt.datetime.utcnow() - dt.timedelta(minutes=240)),
|
||||
unix_time_millis(dt.datetime.utcnow() - dt.timedelta(minutes=180)),
|
||||
unix_time_millis(dt.datetime.utcnow() - dt.timedelta(minutes=120))]
|
||||
'time': [unix_time_millis(utcnow() - dt.timedelta(minutes=240)),
|
||||
unix_time_millis(utcnow() - dt.timedelta(minutes=180)),
|
||||
unix_time_millis(utcnow() - dt.timedelta(minutes=120))]
|
||||
})
|
||||
result = query_satisfied(start_datetime, records, 60)
|
||||
if result is None:
|
||||
|
|
@ -132,7 +137,7 @@ class TestSharedUtilities(unittest.TestCase):
|
|||
|
||||
def test_ts_of_n_minutes_ago(self):
|
||||
print('Testing ts_of_n_minutes_ago()')
|
||||
now = dt.datetime.utcnow()
|
||||
now = utcnow()
|
||||
|
||||
test_cases = [
|
||||
(60, 1), # 60 candles of 1 minute each
|
||||
|
|
|
|||
|
|
@ -0,0 +1,403 @@
|
|||
"""
|
||||
Tests for the strategy generation pipeline.
|
||||
Tests the flow: AI description → Blockly XML → JSON → Python
|
||||
"""
|
||||
import json
|
||||
import pytest
|
||||
import subprocess
|
||||
import xml.etree.ElementTree as ET
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add src to path for imports
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
|
||||
|
||||
from PythonGenerator import PythonGenerator
|
||||
|
||||
|
||||
class TestStrategyBuilder:
|
||||
"""Tests for the CmdForge strategy-builder tool."""
|
||||
|
||||
@staticmethod
|
||||
def _get_blocks(root):
|
||||
"""Get all block elements, handling XML namespaces."""
|
||||
# Blockly uses namespace https://developers.google.com/blockly/xml
|
||||
# Elements may be prefixed with {namespace}block or just block
|
||||
blocks = []
|
||||
for elem in root.iter():
|
||||
# Get local name without namespace
|
||||
tag = elem.tag.split('}')[-1] if '}' in elem.tag else elem.tag
|
||||
if tag == 'block':
|
||||
blocks.append(elem)
|
||||
return blocks
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_simple_rsi_strategy(self):
|
||||
"""Test generating a simple RSI-based strategy."""
|
||||
input_data = {
|
||||
"description": "Buy when RSI is below 30 and sell when RSI is above 70",
|
||||
"indicators": [{"name": "RSI", "outputs": ["RSI"]}],
|
||||
"signals": [],
|
||||
"default_source": {"exchange": "binance", "market": "BTC/USDT", "timeframe": "5m"}
|
||||
}
|
||||
|
||||
result = subprocess.run(
|
||||
['strategy-builder'],
|
||||
input=json.dumps(input_data),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120
|
||||
)
|
||||
|
||||
assert result.returncode == 0, f"Strategy builder failed: {result.stderr}"
|
||||
|
||||
# Validate XML structure
|
||||
xml_output = result.stdout.strip()
|
||||
root = ET.fromstring(xml_output)
|
||||
|
||||
# Check it has the required structure
|
||||
root_tag = root.tag.split('}')[-1] if '}' in root.tag else root.tag
|
||||
assert root_tag == 'xml', f"Root should be 'xml', got {root_tag}"
|
||||
|
||||
blocks = self._get_blocks(root)
|
||||
assert len(blocks) >= 2, f"Should have at least 2 blocks (buy and sell), got {len(blocks)}"
|
||||
|
||||
# Check for execute_if blocks
|
||||
execute_if_blocks = [b for b in blocks if b.get('type') == 'execute_if']
|
||||
assert len(execute_if_blocks) >= 2, "Should have at least 2 execute_if blocks"
|
||||
|
||||
# Check for trade_action blocks
|
||||
trade_action_blocks = [b for b in blocks if b.get('type') == 'trade_action']
|
||||
assert len(trade_action_blocks) >= 2, "Should have buy and sell trade actions"
|
||||
|
||||
# Check for indicator blocks
|
||||
indicator_blocks = [b for b in blocks if 'indicator_RSI' in (b.get('type') or '')]
|
||||
assert len(indicator_blocks) >= 2, "Should use RSI indicator"
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_ema_crossover_strategy(self):
|
||||
"""Test generating an EMA crossover strategy."""
|
||||
input_data = {
|
||||
"description": "Buy when EMA 20 crosses above EMA 50, sell when EMA 20 crosses below EMA 50",
|
||||
"indicators": [
|
||||
{"name": "EMA_20", "outputs": ["ema"]},
|
||||
{"name": "EMA_50", "outputs": ["ema"]}
|
||||
],
|
||||
"signals": [],
|
||||
"default_source": {"exchange": "binance", "market": "ETH/USDT", "timeframe": "1h"}
|
||||
}
|
||||
|
||||
result = subprocess.run(
|
||||
['strategy-builder'],
|
||||
input=json.dumps(input_data),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120
|
||||
)
|
||||
|
||||
assert result.returncode == 0, f"Strategy builder failed: {result.stderr}"
|
||||
|
||||
xml_output = result.stdout.strip()
|
||||
root = ET.fromstring(xml_output)
|
||||
|
||||
# Check for EMA indicator blocks
|
||||
blocks = self._get_blocks(root)
|
||||
ema_20_blocks = [b for b in blocks if 'indicator_EMA_20' in (b.get('type') or '')]
|
||||
ema_50_blocks = [b for b in blocks if 'indicator_EMA_50' in (b.get('type') or '')]
|
||||
|
||||
assert len(ema_20_blocks) >= 1, "Should use EMA_20 indicator"
|
||||
assert len(ema_50_blocks) >= 1, "Should use EMA_50 indicator"
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_no_indicators_price_only(self):
|
||||
"""Test generating a price-based strategy without indicators."""
|
||||
input_data = {
|
||||
"description": "Buy when price drops 5% from previous candle, sell when price rises 3%",
|
||||
"indicators": [],
|
||||
"signals": [],
|
||||
"default_source": {"exchange": "binance", "market": "BTC/USDT", "timeframe": "5m"}
|
||||
}
|
||||
|
||||
result = subprocess.run(
|
||||
['strategy-builder'],
|
||||
input=json.dumps(input_data),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120
|
||||
)
|
||||
|
||||
assert result.returncode == 0, f"Strategy builder failed: {result.stderr}"
|
||||
|
||||
xml_output = result.stdout.strip()
|
||||
root = ET.fromstring(xml_output)
|
||||
|
||||
# Should be valid XML
|
||||
assert root is not None
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_missing_indicator_error(self):
|
||||
"""Test that strategy mentioning indicators without config fails."""
|
||||
input_data = {
|
||||
"description": "Buy when RSI is below 30",
|
||||
"indicators": [], # No indicators configured
|
||||
"signals": [],
|
||||
"default_source": {"exchange": "binance", "market": "BTC/USDT", "timeframe": "5m"}
|
||||
}
|
||||
|
||||
result = subprocess.run(
|
||||
['strategy-builder'],
|
||||
input=json.dumps(input_data),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120
|
||||
)
|
||||
|
||||
assert result.returncode != 0, "Should fail when indicators mentioned but not configured"
|
||||
assert "indicator" in result.stderr.lower(), "Error should mention indicators"
|
||||
|
||||
|
||||
class TestPythonGenerator:
|
||||
"""Tests for the PythonGenerator class."""
|
||||
|
||||
def test_simple_execute_if(self):
|
||||
"""Test generating Python from a simple execute_if block."""
|
||||
strategy_json = {
|
||||
"type": "strategy",
|
||||
"statements": [
|
||||
{
|
||||
"type": "execute_if",
|
||||
"inputs": {
|
||||
"CONDITION": {
|
||||
"type": "comparison",
|
||||
"operator": ">",
|
||||
"inputs": {
|
||||
"LEFT": {"type": "current_price"},
|
||||
"RIGHT": {"type": "dynamic_value", "values": [50000]}
|
||||
}
|
||||
}
|
||||
},
|
||||
"statements": {
|
||||
"DO": [
|
||||
{
|
||||
"type": "trade_action",
|
||||
"trade_type": "buy",
|
||||
"inputs": {"size": 0.01}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
generator = PythonGenerator(
|
||||
default_source={"exchange": "binance", "market": "BTC/USDT", "timeframe": "5m"},
|
||||
strategy_id="test_strategy"
|
||||
)
|
||||
|
||||
result = generator.generate(strategy_json)
|
||||
|
||||
assert "generated_code" in result
|
||||
code = result["generated_code"]
|
||||
|
||||
# Check for expected code elements
|
||||
assert "def next():" in code
|
||||
assert "if " in code
|
||||
assert "get_current_price" in code
|
||||
assert "trade_order" in code
|
||||
|
||||
def test_indicator_condition(self):
|
||||
"""Test generating Python with indicator conditions."""
|
||||
strategy_json = {
|
||||
"type": "strategy",
|
||||
"statements": [
|
||||
{
|
||||
"type": "execute_if",
|
||||
"inputs": {
|
||||
"CONDITION": {
|
||||
"type": "comparison",
|
||||
"operator": "<",
|
||||
"inputs": {
|
||||
"LEFT": {
|
||||
"type": "indicator_RSI",
|
||||
"fields": {"OUTPUT": "RSI"}
|
||||
},
|
||||
"RIGHT": {"type": "dynamic_value", "values": [30]}
|
||||
}
|
||||
}
|
||||
},
|
||||
"statements": {
|
||||
"DO": [
|
||||
{
|
||||
"type": "trade_action",
|
||||
"trade_type": "buy",
|
||||
"inputs": {"size": 0.1}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
generator = PythonGenerator(
|
||||
default_source={"exchange": "binance", "market": "BTC/USDT", "timeframe": "5m"},
|
||||
strategy_id="test_indicator"
|
||||
)
|
||||
|
||||
result = generator.generate(strategy_json)
|
||||
code = result["generated_code"]
|
||||
|
||||
# Check for indicator processing
|
||||
assert "process_indicator" in code
|
||||
assert "RSI" in code
|
||||
|
||||
# Check indicators are tracked
|
||||
assert len(result["indicators"]) > 0
|
||||
|
||||
def test_logical_and_condition(self):
|
||||
"""Test generating Python with logical AND conditions."""
|
||||
strategy_json = {
|
||||
"type": "strategy",
|
||||
"statements": [
|
||||
{
|
||||
"type": "execute_if",
|
||||
"inputs": {
|
||||
"CONDITION": {
|
||||
"type": "logical_and",
|
||||
"inputs": {
|
||||
"left": {
|
||||
"type": "comparison",
|
||||
"operator": "<",
|
||||
"inputs": {
|
||||
"LEFT": {"type": "indicator_RSI", "fields": {"OUTPUT": "RSI"}},
|
||||
"RIGHT": {"type": "dynamic_value", "values": [30]}
|
||||
}
|
||||
},
|
||||
"right": {
|
||||
"type": "flag_is_set",
|
||||
"flag_name": "bought",
|
||||
"flag_value": False
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"statements": {
|
||||
"DO": [
|
||||
{"type": "trade_action", "trade_type": "buy", "inputs": {"size": 0.01}}
|
||||
]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
generator = PythonGenerator(
|
||||
default_source={"exchange": "binance", "market": "BTC/USDT", "timeframe": "5m"},
|
||||
strategy_id="test_and"
|
||||
)
|
||||
|
||||
result = generator.generate(strategy_json)
|
||||
code = result["generated_code"]
|
||||
|
||||
# Check for logical AND
|
||||
assert " and " in code
|
||||
assert "flags.get" in code
|
||||
|
||||
def test_set_flag(self):
|
||||
"""Test generating Python for flag setting."""
|
||||
strategy_json = {
|
||||
"type": "strategy",
|
||||
"statements": [
|
||||
{
|
||||
"type": "set_flag",
|
||||
"flag_name": "bought",
|
||||
"flag_value": "True"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
generator = PythonGenerator(
|
||||
default_source={"exchange": "binance", "market": "BTC/USDT", "timeframe": "5m"},
|
||||
strategy_id="test_flag"
|
||||
)
|
||||
|
||||
result = generator.generate(strategy_json)
|
||||
code = result["generated_code"]
|
||||
|
||||
# Check for flag setting
|
||||
assert "flags['bought']" in code
|
||||
assert "True" in code
|
||||
|
||||
# Check flag is tracked
|
||||
assert "bought" in result["flags_used"]
|
||||
|
||||
def test_trade_action_with_options(self):
|
||||
"""Test generating Python for trade with stop loss and take profit."""
|
||||
strategy_json = {
|
||||
"type": "strategy",
|
||||
"statements": [
|
||||
{
|
||||
"type": "trade_action",
|
||||
"trade_type": "buy",
|
||||
"inputs": {"size": 0.1},
|
||||
"trade_options": [
|
||||
{"type": "stop_loss", "inputs": {"stop_loss": 45000}},
|
||||
{"type": "take_profit", "inputs": {"take_profit": 55000}}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
generator = PythonGenerator(
|
||||
default_source={"exchange": "binance", "market": "BTC/USDT", "timeframe": "5m"},
|
||||
strategy_id="test_trade_options"
|
||||
)
|
||||
|
||||
result = generator.generate(strategy_json)
|
||||
code = result["generated_code"]
|
||||
|
||||
# Check for trade order call
|
||||
assert "trade_order" in code
|
||||
assert "buy" in code
|
||||
|
||||
def test_math_operation(self):
|
||||
"""Test generating Python for math operations."""
|
||||
strategy_json = {
|
||||
"type": "strategy",
|
||||
"statements": [
|
||||
{
|
||||
"type": "execute_if",
|
||||
"inputs": {
|
||||
"CONDITION": {
|
||||
"type": "comparison",
|
||||
"operator": ">",
|
||||
"inputs": {
|
||||
"LEFT": {"type": "current_price"},
|
||||
"RIGHT": {
|
||||
"type": "math_operation",
|
||||
"inputs": {
|
||||
"operator": "MULTIPLY",
|
||||
"left_operand": {"type": "indicator_SMA", "fields": {"OUTPUT": "sma"}},
|
||||
"right_operand": {"type": "dynamic_value", "values": [1.02]}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"statements": {
|
||||
"DO": [
|
||||
{"type": "trade_action", "trade_type": "sell", "inputs": {"size": 0.01}}
|
||||
]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
generator = PythonGenerator(
|
||||
default_source={"exchange": "binance", "market": "BTC/USDT", "timeframe": "5m"},
|
||||
strategy_id="test_math"
|
||||
)
|
||||
|
||||
result = generator.generate(strategy_json)
|
||||
code = result["generated_code"]
|
||||
|
||||
# Check for math operation
|
||||
assert "*" in code # Multiply operator
|
||||
|
|
@ -232,7 +232,7 @@ class TestTrades:
|
|||
)
|
||||
|
||||
assert status == 'Error'
|
||||
assert 'No exchange connected' in msg
|
||||
assert 'No exchange' in msg.lower() or 'no exchange' in msg.lower()
|
||||
|
||||
def test_get_trades_json(self, mock_users):
|
||||
"""Test getting trades in JSON format."""
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import pytest
|
||||
from trade import Trade
|
||||
|
||||
|
||||
|
|
@ -82,13 +83,13 @@ def test_update_values():
|
|||
assert position_size == 5
|
||||
pl = trade_obj.get_pl()
|
||||
print(f'PL reported: {pl}')
|
||||
# Should be: - 1/2 * opening value - opening value * fee - closing value * fee
|
||||
# 5 - 1 - 0.5 = -6.5
|
||||
assert pl == -6.5
|
||||
# With 0.1% fee (0.001): gross loss -5, fees = (5*0.001) + (10*0.001) = 0.015
|
||||
# Net PL: -5 - 0.015 = -5.015
|
||||
assert pl == pytest.approx(-5.015)
|
||||
pl_pct = trade_obj.get_pl_pct()
|
||||
print(f'PL% reported: {pl_pct}')
|
||||
# Should be -6.5/10 = -65%
|
||||
assert pl_pct == -65
|
||||
# Should be -5.015/10 * 100 = -50.15%
|
||||
assert pl_pct == pytest.approx(-50.15)
|
||||
|
||||
# Add 1/2 to the price of the quote symbol.
|
||||
current_price = 150
|
||||
|
|
@ -100,13 +101,13 @@ def test_update_values():
|
|||
assert position_size == 15
|
||||
pl = trade_obj.get_pl()
|
||||
print(f'PL reported: {pl}')
|
||||
# Should be 5 - opening fee - closing fee
|
||||
# fee should be (10 * .1) + (15 * .1) = 2.5
|
||||
assert pl == 2.5
|
||||
# With 0.1% fee (0.001): gross profit 5, fees = (15*0.001) + (10*0.001) = 0.025
|
||||
# Net PL: 5 - 0.025 = 4.975
|
||||
assert pl == pytest.approx(4.975)
|
||||
pl_pct = trade_obj.get_pl_pct()
|
||||
print(f'PL% reported: {pl_pct}')
|
||||
# should be 2.5/10 = 25%
|
||||
assert pl_pct == 25
|
||||
# Should be 4.975/10 * 100 = 49.75%
|
||||
assert pl_pct == pytest.approx(49.75)
|
||||
|
||||
|
||||
def test_update():
|
||||
|
|
@ -121,7 +122,7 @@ def test_update():
|
|||
current_price = 50
|
||||
result = trade_obj.update(current_price)
|
||||
print(f'The result {result}')
|
||||
assert result == 'inactive'
|
||||
assert result == 'updated' # update() returns 'updated' for inactive trades
|
||||
|
||||
# Simulate a placed trade.
|
||||
trade_obj.trade_filled(0.01, 1000)
|
||||
|
|
@ -150,7 +151,7 @@ def test_trade_filled():
|
|||
trade_obj.trade_filled(qty=0.05, price=100)
|
||||
status = trade_obj.get_status()
|
||||
print(f'\n Status after trade_filled() called: {status}')
|
||||
assert status == 'part_filled'
|
||||
assert status == 'part-filled'
|
||||
|
||||
trade_obj.trade_filled(qty=0.05, price=100)
|
||||
status = trade_obj.get_status()
|
||||
|
|
|
|||
Loading…
Reference in New Issue