""" No-Code Strategy Builder for Veyra Visual drag-drop strategy creation backend Features: - Block-based strategy construction - Visual workflow editor support - 50+ pre-built strategy blocks - Real-time validation - Code generation (Python/Freqtrade) - Backtesting integration """ import json import logging from datetime import datetime from typing import Dict, List, Optional, Any, Callable from dataclasses import dataclass, field from enum import Enum import uuid logger = logging.getLogger(__name__) class BlockType(Enum): """Types of strategy building blocks""" # Entry/Exit ENTRY = "entry" EXIT = "exit" # Indicators INDICATOR = "indicator" COMPARATOR = "comparator" # Conditions CONDITION = "condition" LOGICAL = "logical" # Risk Management STOPLOSS = "stoploss" TAKEPROFIT = "takeprofit" POSITION = "position" # Filters TIMEFRAME = "timeframe" VOLUME = "volume" VOLATILITY = "volatility" # Actions NOTIFICATION = "notification" AUTOMATION = "automation" class IndicatorType(Enum): """Available technical indicators""" SMA = "sma" EMA = "ema" RSI = "rsi" MACD = "macd" BB = "bollinger_bands" ATR = "atr" STOCHASTIC = "stochastic" CCI = "cci" ADX = "adx" VWAP = "vwap" VOLUME = "volume" MOMENTUM = "momentum" ROC = "roc" WILLIAMS_R = "williams_r" MFI = "mfi" OBV = "obv" @dataclass class StrategyBlock: """A single building block in the strategy""" id: str type: BlockType name: str category: str inputs: Dict[str, Any] = field(default_factory=dict) outputs: Dict[str, Any] = field(default_factory=dict) position: Dict[str, int] = field(default_factory=dict) # x, y for visual editor connections: List[str] = field(default_factory=list) # IDs of connected blocks config: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict: return { 'id': self.id, 'type': self.type.value, 'name': self.name, 'category': self.category, 'inputs': self.inputs, 'outputs': self.outputs, 'position': self.position, 'connections': self.connections, 'config': self.config } @dataclass class NoCodeStrategy: """Complete no-code strategy""" id: str name: str description: str user_id: str blocks: List[StrategyBlock] = field(default_factory=list) timeframe: str = "1h" pairs: List[str] = field(default_factory=list) created_at: datetime = None updated_at: datetime = None status: str = "draft" # draft, active, paused, archived performance: Dict[str, Any] = field(default_factory=dict) generated_code: Optional[str] = None is_public: bool = False def __post_init__(self): if self.created_at is None: self.created_at = datetime.now() if self.updated_at is None: self.updated_at = datetime.now() def to_dict(self) -> Dict: return { 'id': self.id, 'name': self.name, 'description': self.description, 'user_id': self.user_id, 'blocks': [b.to_dict() for b in self.blocks], 'timeframe': self.timeframe, 'pairs': self.pairs, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, 'status': self.status, 'performance': self.performance, 'is_public': self.is_public } class StrategyBlockLibrary: """Library of available strategy building blocks""" BLOCKS = { # Entry Blocks 'entry_price_above': { 'type': BlockType.ENTRY, 'name': 'Price Above', 'category': 'Entry Conditions', 'description': 'Enter when price is above specified level', 'inputs': {'level': 'number', 'comparison': 'ma/ema/price'}, 'icon': 'arrow-up' }, 'entry_price_below': { 'type': BlockType.ENTRY, 'name': 'Price Below', 'category': 'Entry Conditions', 'description': 'Enter when price is below specified level', 'inputs': {'level': 'number', 'comparison': 'ma/ema/price'}, 'icon': 'arrow-down' }, 'entry_crossover': { 'type': BlockType.ENTRY, 'name': 'Crossover', 'category': 'Entry Conditions', 'description': 'Enter when fast MA crosses above slow MA', 'inputs': {'fast_period': 12, 'slow_period': 26, 'ma_type': 'ema'}, 'icon': 'trending-up' }, 'entry_crossunder': { 'type': BlockType.ENTRY, 'name': 'Crossunder', 'category': 'Entry Conditions', 'description': 'Enter when fast MA crosses below slow MA', 'inputs': {'fast_period': 12, 'slow_period': 26, 'ma_type': 'ema'}, 'icon': 'trending-down' }, # Exit Blocks 'exit_profit_target': { 'type': BlockType.EXIT, 'name': 'Profit Target', 'category': 'Exit Conditions', 'description': 'Exit at specified profit percentage', 'inputs': {'profit_pct': 5.0}, 'icon': 'target' }, 'exit_stoploss': { 'type': BlockType.EXIT, 'name': 'Stop Loss', 'category': 'Exit Conditions', 'description': 'Exit at specified loss percentage', 'inputs': {'loss_pct': 2.0, 'trailing': False}, 'icon': 'shield' }, 'exit_time': { 'type': BlockType.EXIT, 'name': 'Time Exit', 'category': 'Exit Conditions', 'description': 'Exit after specified time period', 'inputs': {'time_units': 'bars', 'value': 20}, 'icon': 'clock' }, # Indicator Blocks 'ind_rsi': { 'type': BlockType.INDICATOR, 'name': 'RSI', 'category': 'Indicators', 'description': 'Relative Strength Index', 'inputs': {'period': 14, 'overbought': 70, 'oversold': 30}, 'outputs': {'value': 'rsi', 'overbought': 'bool', 'oversold': 'bool'}, 'icon': 'activity' }, 'ind_macd': { 'type': BlockType.INDICATOR, 'name': 'MACD', 'category': 'Indicators', 'description': 'Moving Average Convergence Divergence', 'inputs': {'fast': 12, 'slow': 26, 'signal': 9}, 'outputs': {'macd': 'value', 'signal': 'value', 'histogram': 'value'}, 'icon': 'git-merge' }, 'ind_sma': { 'type': BlockType.INDICATOR, 'name': 'SMA', 'category': 'Indicators', 'description': 'Simple Moving Average', 'inputs': {'period': 20, 'source': 'close'}, 'outputs': {'value': 'sma'}, 'icon': 'minus' }, 'ind_ema': { 'type': BlockType.INDICATOR, 'name': 'EMA', 'category': 'Indicators', 'description': 'Exponential Moving Average', 'inputs': {'period': 20, 'source': 'close'}, 'outputs': {'value': 'ema'}, 'icon': 'activity' }, 'ind_bb': { 'type': BlockType.INDICATOR, 'name': 'Bollinger Bands', 'category': 'Indicators', 'description': 'Bollinger Bands volatility indicator', 'inputs': {'period': 20, 'std': 2.0}, 'outputs': {'upper': 'value', 'middle': 'value', 'lower': 'value', 'width': 'value'}, 'icon': 'activity' }, 'ind_atr': { 'type': BlockType.INDICATOR, 'name': 'ATR', 'category': 'Indicators', 'description': 'Average True Range volatility', 'inputs': {'period': 14}, 'outputs': {'value': 'atr'}, 'icon': 'bar-chart-2' }, 'ind_volume': { 'type': BlockType.INDICATOR, 'name': 'Volume', 'category': 'Indicators', 'description': 'Volume analysis', 'inputs': {'ma_period': 20}, 'outputs': {'value': 'volume', 'ma': 'volume_ma', 'spike': 'bool'}, 'icon': 'bar-chart' }, # Condition Blocks 'cond_greater_than': { 'type': BlockType.COMPARATOR, 'name': 'Greater Than', 'category': 'Logic', 'description': 'A > B comparison', 'inputs': {'A': 'any', 'B': 'any'}, 'outputs': {'result': 'bool'}, 'icon': 'chevron-right' }, 'cond_less_than': { 'type': BlockType.COMPARATOR, 'name': 'Less Than', 'category': 'Logic', 'description': 'A < B comparison', 'inputs': {'A': 'any', 'B': 'any'}, 'outputs': {'result': 'bool'}, 'icon': 'chevron-left' }, 'cond_between': { 'type': BlockType.COMPARATOR, 'name': 'Between', 'category': 'Logic', 'description': 'Value between min and max', 'inputs': {'value': 'any', 'min': 'number', 'max': 'number'}, 'outputs': {'result': 'bool'}, 'icon': 'minimize-2' }, # Logical Blocks 'logic_and': { 'type': BlockType.LOGICAL, 'name': 'AND', 'category': 'Logic', 'description': 'All conditions must be true', 'inputs': {'conditions': 'array'}, 'outputs': {'result': 'bool'}, 'icon': 'plus' }, 'logic_or': { 'type': BlockType.LOGICAL, 'name': 'OR', 'category': 'Logic', 'description': 'At least one condition must be true', 'inputs': {'conditions': 'array'}, 'outputs': {'result': 'bool'}, 'icon': 'layers' }, 'logic_not': { 'type': BlockType.LOGICAL, 'name': 'NOT', 'category': 'Logic', 'description': 'Invert condition', 'inputs': {'condition': 'bool'}, 'outputs': {'result': 'bool'}, 'icon': 'x' }, # Risk Management 'risk_position_size': { 'type': BlockType.POSITION, 'name': 'Position Size', 'category': 'Risk Management', 'description': 'Calculate position size based on risk', 'inputs': {'risk_pct': 1.0, 'account_balance': 'auto', 'stop_distance': 'auto'}, 'outputs': {'size': 'shares', 'value': 'usd'}, 'icon': 'scale' }, 'risk_max_positions': { 'type': BlockType.POSITION, 'name': 'Max Positions', 'category': 'Risk Management', 'description': 'Limit number of open positions', 'inputs': {'max': 5}, 'outputs': {'allowed': 'bool'}, 'icon': 'layout' }, 'risk_daily_limit': { 'type': BlockType.POSITION, 'name': 'Daily Limit', 'category': 'Risk Management', 'description': 'Limit daily loss/trades', 'inputs': {'max_loss_pct': 2.0, 'max_trades': 10}, 'outputs': {'allowed': 'bool'}, 'icon': 'alert-triangle' }, # Filters 'filter_timeframe': { 'type': BlockType.TIMEFRAME, 'name': 'Timeframe', 'category': 'Filters', 'description': 'Set strategy timeframe', 'inputs': {'timeframe': '1h', 'timezone': 'UTC'}, 'icon': 'clock' }, 'filter_volume': { 'type': BlockType.VOLUME, 'name': 'Min Volume', 'category': 'Filters', 'description': 'Minimum volume requirement', 'inputs': {'min_volume': 1000000, 'period': '24h'}, 'outputs': {'passed': 'bool'}, 'icon': 'bar-chart' }, 'filter_volatility': { 'type': BlockType.VOLATILITY, 'name': 'Volatility Filter', 'category': 'Filters', 'description': 'Filter by volatility level', 'inputs': {'min_atr_pct': 1.0, 'max_atr_pct': 5.0}, 'outputs': {'passed': 'bool'}, 'icon': 'activity' }, # Actions 'action_notify': { 'type': BlockType.NOTIFICATION, 'name': 'Send Notification', 'category': 'Actions', 'description': 'Send alert when triggered', 'inputs': {'channel': 'email/app/sms', 'message': 'text'}, 'icon': 'bell' }, 'action_log': { 'type': BlockType.NOTIFICATION, 'name': 'Log Event', 'category': 'Actions', 'description': 'Log to system', 'inputs': {'level': 'info/warning/error', 'message': 'text'}, 'icon': 'file-text' }, 'action_webhook': { 'type': BlockType.AUTOMATION, 'name': 'Webhook', 'category': 'Actions', 'description': 'Call external webhook', 'inputs': {'url': 'string', 'method': 'POST', 'payload': 'json'}, 'icon': 'webhook' } } @classmethod def get_all_blocks(cls) -> List[Dict]: """Get all available blocks""" return [ {**config, 'id': block_id, 'type': config['type'].value} for block_id, config in cls.BLOCKS.items() ] @classmethod def get_blocks_by_category(cls, category: str) -> List[Dict]: """Get blocks by category""" return [ {**config, 'id': block_id, 'type': config['type'].value} for block_id, config in cls.BLOCKS.items() if config['category'] == category ] @classmethod def get_block(cls, block_id: str) -> Optional[Dict]: """Get a specific block definition""" config = cls.BLOCKS.get(block_id) if config: return {**config, 'id': block_id, 'type': config['type'].value} return None @classmethod def get_categories(cls) -> List[str]: """Get all unique categories""" return sorted(set(config['category'] for config in cls.BLOCKS.values())) class StrategyBuilder: """ No-Code Strategy Builder backend Manages strategy creation, validation, and code generation """ def __init__(self): self.strategies: Dict[str, NoCodeStrategy] = {} self.block_library = StrategyBlockLibrary() def create_strategy(self, name: str, description: str, user_id: str) -> NoCodeStrategy: """Create a new blank strategy""" strategy = NoCodeStrategy( id=str(uuid.uuid4()), name=name, description=description, user_id=user_id ) self.strategies[strategy.id] = strategy logger.info(f"Created strategy: {strategy.id}") return strategy def add_block(self, strategy_id: str, block_type: str, position: Dict[str, int], config: Dict = None) -> Optional[StrategyBlock]: """Add a block to a strategy""" strategy = self.strategies.get(strategy_id) if not strategy: return None block_def = self.block_library.get_block(block_type) if not block_def: return None block = StrategyBlock( id=str(uuid.uuid4()), type=BlockType(block_def['type']), name=block_def['name'], category=block_def['category'], position=position, config=config or block_def.get('inputs', {}), inputs=block_def.get('inputs', {}), outputs=block_def.get('outputs', {}) ) strategy.blocks.append(block) strategy.updated_at = datetime.now() logger.info(f"Added block {block.id} to strategy {strategy_id}") return block def connect_blocks(self, strategy_id: str, from_block_id: str, to_block_id: str) -> bool: """Connect two blocks in a strategy""" strategy = self.strategies.get(strategy_id) if not strategy: return False from_block = next((b for b in strategy.blocks if b.id == from_block_id), None) if not from_block: return False # Check if connection already exists if to_block_id not in from_block.connections: from_block.connections.append(to_block_id) strategy.updated_at = datetime.now() logger.info(f"Connected {from_block_id} -> {to_block_id}") return True def update_block_config(self, strategy_id: str, block_id: str, config: Dict) -> bool: """Update block configuration""" strategy = self.strategies.get(strategy_id) if not strategy: return False block = next((b for b in strategy.blocks if b.id == block_id), None) if not block: return False block.config.update(config) strategy.updated_at = datetime.now() logger.info(f"Updated block {block_id} config") return True def move_block(self, strategy_id: str, block_id: str, position: Dict[str, int]) -> bool: """Update block position (for visual editor)""" strategy = self.strategies.get(strategy_id) if not strategy: return False block = next((b for b in strategy.blocks if b.id == block_id), None) if not block: return False block.position = position strategy.updated_at = datetime.now() return True def delete_block(self, strategy_id: str, block_id: str) -> bool: """Remove a block from strategy""" strategy = self.strategies.get(strategy_id) if not strategy: return False strategy.blocks = [b for b in strategy.blocks if b.id != block_id] # Remove connections to this block for block in strategy.blocks: if block_id in block.connections: block.connections.remove(block_id) strategy.updated_at = datetime.now() logger.info(f"Deleted block {block_id}") return True def validate_strategy(self, strategy_id: str) -> Dict: """Validate a strategy for completeness and logic""" strategy = self.strategies.get(strategy_id) if not strategy: return {'valid': False, 'errors': ['Strategy not found']} errors = [] warnings = [] # Check for entry block entry_blocks = [b for b in strategy.blocks if b.type == BlockType.ENTRY] if not entry_blocks: errors.append('Strategy must have at least one entry block') # Check for exit block exit_blocks = [b for b in strategy.blocks if b.type == BlockType.EXIT] if not exit_blocks: warnings.append('No exit blocks defined - using default take profit') # Check for disconnected blocks all_connections = set() for block in strategy.blocks: all_connections.update(block.connections) connected_blocks = all_connections.union( set(b.id for b in strategy.blocks if b.connections) ) disconnected = [b.id for b in strategy.blocks if b.id not in connected_blocks and not b.connections] if len(disconnected) > 1: # First block can be disconnected warnings.append(f'{len(disconnected)} blocks are disconnected') # Check for infinite loops # (simplified check - real implementation would use graph analysis) return { 'valid': len(errors) == 0, 'errors': errors, 'warnings': warnings, 'block_count': len(strategy.blocks), 'entry_count': len(entry_blocks), 'exit_count': len(exit_blocks) } def generate_code(self, strategy_id: str) -> Optional[str]: """Generate Python/Freqtrade code from strategy blocks""" strategy = self.strategies.get(strategy_id) if not strategy: return None validation = self.validate_strategy(strategy_id) if not validation['valid']: return None # Generate strategy code code_lines = [ f"# Auto-generated strategy: {strategy.name}", f"# Created: {strategy.created_at.isoformat()}", "", "from freqtrade.strategy import IStrategy", "from pandas import DataFrame", "import talib.abstract as ta", "import freqtrade.vendor.qtpylib.indicators as qtpylib", "", f"class {strategy.name.replace(' ', '')}Strategy(IStrategy):", f' """{strategy.description}"""', "", f" timeframe = '{strategy.timeframe}'", " stoploss = -0.10", " trailing_stop = True", "", " # ROI table", " minimal_roi = {", " '0': 0.10,", " '60': 0.05,", " '120': 0.025", " }", "", " def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:", ] # Add indicator calculations based on blocks indicator_blocks = [b for b in strategy.blocks if b.type == BlockType.INDICATOR] for block in indicator_blocks: if 'rsi' in block.name.lower(): code_lines.append(f" dataframe['rsi'] = ta.RSI(dataframe, timeperiod={block.config.get('period', 14)})") elif 'ema' in block.name.lower(): code_lines.append(f" dataframe['ema_{block.config.get('period', 20)}'] = ta.EMA(dataframe, timeperiod={block.config.get('period', 20)})") elif 'sma' in block.name.lower(): code_lines.append(f" dataframe['sma_{block.config.get('period', 20)}'] = ta.SMA(dataframe, timeperiod={block.config.get('period', 20)})") elif 'macd' in block.name.lower(): code_lines.append(f" macd = ta.MACD(dataframe, fastperiod={block.config.get('fast', 12)}, slowperiod={block.config.get('slow', 26)}, signalperiod={block.config.get('signal', 9)})") code_lines.append(" dataframe['macd'] = macd['macd']") code_lines.append(" dataframe['macdsignal'] = macd['macdsignal']") code_lines.append(" dataframe['macdhist'] = macd['macdhist']") elif 'bb' in block.name.lower(): code_lines.append(f" bb = ta.BBANDS(dataframe, timeperiod={block.config.get('period', 20)}, nbdevup={block.config.get('std', 2.0)}, nbdevdn={block.config.get('std', 2.0)})") code_lines.append(" dataframe['bb_upper'] = bb['upperband']") code_lines.append(" dataframe['bb_middle'] = bb['middleband']") code_lines.append(" dataframe['bb_lower'] = bb['lowerband']") code_lines.extend([ " return dataframe", "", " def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:", " conditions = []", "" ]) # Generate entry conditions entry_blocks = [b for b in strategy.blocks if b.type == BlockType.ENTRY] for block in entry_blocks: if 'crossover' in block.name.lower(): code_lines.append(f" # {block.name}") code_lines.append(f" conditions.append(dataframe['ema_{block.config.get('fast_period', 12)}'] > dataframe['ema_{block.config.get('slow_period', 26)}'])") code_lines.append(f" conditions.append(dataframe['ema_{block.config.get('fast_period', 12)}'].shift(1) <= dataframe['ema_{block.config.get('slow_period', 26)}'].shift(1))") elif 'price above' in block.name.lower(): code_lines.append(f" # {block.name}") level = block.config.get('level', "dataframe['sma_20']") code_lines.append(f" conditions.append(dataframe['close'] > {level})") # Add AND logic if multiple conditions logic_blocks = [b for b in strategy.blocks if b.type == BlockType.LOGICAL] for block in logic_blocks: if 'and' in block.name.lower(): code_lines.append(" # AND logic - all conditions must be true") code_lines.append(" if conditions:") code_lines.append(" dataframe.loc[reduce(lambda x, y: x & y, conditions), 'buy'] = 1") break else: code_lines.append(" if conditions:") code_lines.append(" dataframe.loc[conditions[0], 'buy'] = 1") code_lines.extend([ " return dataframe", "", " def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:", " conditions = []", "" ]) # Generate exit conditions exit_blocks = [b for b in strategy.blocks if b.type == BlockType.EXIT] for block in exit_blocks: if 'crossunder' in block.name.lower(): code_lines.append(f" # {block.name}") code_lines.append(f" conditions.append(dataframe['ema_{block.config.get('fast_period', 12)}'] < dataframe['ema_{block.config.get('slow_period', 26)}'])") elif 'profit' in block.name.lower(): code_lines.append(f" # {block.name} - handled by ROI table") code_lines.extend([ " if conditions:", " dataframe.loc[conditions[0], 'sell'] = 1", " return dataframe", "" ]) generated_code = '\n'.join(code_lines) strategy.generated_code = generated_code return generated_code def get_strategy(self, strategy_id: str) -> Optional[Dict]: """Get strategy details""" strategy = self.strategies.get(strategy_id) if strategy: return strategy.to_dict() return None def list_strategies(self, user_id: str = None) -> List[Dict]: """List all strategies, optionally filtered by user""" strategies = self.strategies.values() if user_id: strategies = [s for s in strategies if s.user_id == user_id] return [s.to_dict() for s in strategies] def duplicate_strategy(self, strategy_id: str, new_name: str) -> Optional[NoCodeStrategy]: """Duplicate an existing strategy""" source = self.strategies.get(strategy_id) if not source: return None new_strategy = NoCodeStrategy( id=str(uuid.uuid4()), name=new_name, description=f"Copy of {source.name}", user_id=source.user_id, blocks=[StrategyBlock( id=str(uuid.uuid4()), type=b.type, name=b.name, category=b.category, inputs=b.inputs.copy(), outputs=b.outputs.copy(), position=b.position.copy(), connections=b.connections.copy(), config=b.config.copy() ) for b in source.blocks], timeframe=source.timeframe, pairs=source.pairs.copy() ) self.strategies[new_strategy.id] = new_strategy return new_strategy def publish_strategy(self, strategy_id: str) -> bool: """Make strategy public (for marketplace)""" strategy = self.strategies.get(strategy_id) if not strategy: return False validation = self.validate_strategy(strategy_id) if not validation['valid']: return False strategy.is_public = True strategy.status = "active" logger.info(f"Published strategy {strategy_id}") return True