Building a Trading DSL: From Grammar to Execution

1IF RSI(14) > 70 AND PRICE > SMA(50) THEN SELL 100
2

1from lark import Lark, Transformer, v_args
2
3grammar = """
4    ?start: rule
5
6    rule: "IF" condition "THEN" action
7
8    ?condition: expression comparison expression
9              | condition "AND" condition -> and_cond
10              | condition "OR" condition  -> or_cond
11              | "(" condition ")"
12
13    ?expression: term
14               | expression "+" term      -> add
15               | expression "-" term      -> sub
16
17    ?term: NUMBER                         -> number
18         | "PRICE"                        -> price
19         | function_call
20
21    function_call: NAME "(" NUMBER ")"
22
23    ?comparison: ">" -> gt
24               | "<" -> lt
25               | "==" -> eq
26               | ">=" -> gte
27               | "<=" -> lte
28
29    action: "BUY" NUMBER                  -> buy
30          | "SELL" NUMBER                 -> sell
31          | "HOLD"                        -> hold
32
33    NAME: /[A-Z]+/
34    %import common.NUMBER
35    %import common.WS
36    %ignore WS
37"""
38
39parser = Lark(grammar, start='start', parser='lalr')
40

1strategy = "IF RSI(14) > 70 THEN SELL 100"
2tree = parser.parse(strategy)
3print(tree.pretty())
4

1rule
2  gt
3    function_call
4      RSI
5      14
6    number  70
7  sell  100
8

1from lark.exceptions import LarkError, UnexpectedInput, UnexpectedToken
2
3def parse_with_errors(strategy_text: str):
4    try:
5        tree = parser.parse(strategy_text)
6        return tree, None
7    except UnexpectedToken as e:
8        error_msg = f"Syntax error at line {e.line}, column {e.column}:\n"
9        error_msg += f"  Unexpected token: '{e.token}'\n"
10        error_msg += f"  Expected one of: {', '.join(e.expected)}"
11        return None, error_msg
12    except UnexpectedInput as e:
13        error_msg = f"Syntax error: {str(e)}"
14        return None, error_msg
15    except LarkError as e:
16        error_msg = f"Parse error: {str(e)}"
17        return None, error_msg
18
19# Test with invalid syntax
20invalid_strategy = "IF RSI(14) > 70 THEN"  # Missing action
21tree, error = parse_with_errors(invalid_strategy)
22if error:
23    print(f"Error: {error}")
24

1class SemanticError(Exception):
2    pass
3
4class SemanticValidator(Transformer):
5    """Validate semantic correctness of the strategy."""
6    
7    VALID_INDICATORS = {'RSI', 'SMA', 'EMA', 'MACD', 'BBANDS'}
8    
9    def function_call(self, items):
10        func_name = items[0].value
11        period = int(items[1])
12        
13        # Validate function name
14        if func_name not in self.VALID_INDICATORS:
15            raise SemanticError(
16                f"Unknown indicator: {func_name}. "
17                f"Valid indicators: {', '.join(self.VALID_INDICATORS)}"
18            )
19        
20        # Validate period range
21        if period < 1 or period > 200:
22            raise SemanticError(
23                f"Invalid period {period} for {func_name}. "
24                f"Period must be between 1 and 200."
25            )
26        
27        return ('function_call', func_name, period)
28    
29    def buy(self, items):
30        quantity = int(items[0])
31        if quantity <= 0:
32            raise SemanticError(f"Buy quantity must be positive, got {quantity}")
33        return ('buy', quantity)
34    
35    def sell(self, items):
36        quantity = int(items[0])
37        if quantity <= 0:
38            raise SemanticError(f"Sell quantity must be positive, got {quantity}")
39        return ('sell', quantity)
40
41# Test semantic validation
42try:
43    tree = parser.parse("IF INVALID(14) > 70 THEN BUY 100")
44    validator = SemanticValidator()
45    validator.transform(tree)
46except SemanticError as e:
47    print(f"Semantic Error: {e}")
48

1from enum import Enum
2from typing import Union
3
4class Type(Enum):
5    NUMBER = "number"
6    BOOLEAN = "boolean"
7    ACTION = "action"
8
9class TypeChecker(Transformer):
10    """Type checker for the DSL."""
11    
12    def number(self, items):
13        return (Type.NUMBER, float(items[0]))
14    
15    def price(self, _):
16        return (Type.NUMBER, 'PRICE')
17    
18    def function_call(self, items):
19        # All indicators return numbers
20        return (Type.NUMBER, ('function_call', items[0].value, int(items[1])))
21    
22    def gt(self, items):
23        left_type, left_val = items[0]
24        right_type, right_val = items[1]
25        
26        if left_type != Type.NUMBER or right_type != Type.NUMBER:
27            raise TypeError(
28                f"Comparison requires numbers, got {left_type} and {right_type}"
29            )
30        
31        return (Type.BOOLEAN, ('gt', left_val, right_val))
32    
33    def lt(self, items):
34        left_type, left_val = items[0]
35        right_type, right_val = items[1]
36        
37        if left_type != Type.NUMBER or right_type != Type.NUMBER:
38            raise TypeError(
39                f"Comparison requires numbers, got {left_type} and {right_type}"
40            )
41        
42        return (Type.BOOLEAN, ('lt', left_val, right_val))
43    
44    def and_cond(self, items):
45        left_type, left_val = items[0]
46        right_type, right_val = items[1]
47        
48        if left_type != Type.BOOLEAN or right_type != Type.BOOLEAN:
49            raise TypeError(
50                f"AND requires booleans, got {left_type} and {right_type}"
51            )
52        
53        return (Type.BOOLEAN, ('and', left_val, right_val))
54    
55    def rule(self, items):
56        cond_type, cond_val = items[0]
57        action_type, action_val = items[1]
58        
59        if cond_type != Type.BOOLEAN:
60            raise TypeError(f"Condition must be boolean, got {cond_type}")
61        
62        if action_type != Type.ACTION:
63            raise TypeError(f"Action required, got {action_type}")
64        
65        return ('rule', cond_val, action_val)
66    
67    def buy(self, items):
68        return (Type.ACTION, ('buy', int(items[0])))
69    
70    def sell(self, items):
71        return (Type.ACTION, ('sell', int(items[0])))
72
73# Test type checking
74try:
75    tree = parser.parse("IF RSI(14) THEN BUY 100")  # Missing comparison
76    type_checker = TypeChecker()
77    type_checker.transform(tree)
78except TypeError as e:
79    print(f"Type Error: {e}")
80

1class StrategyExecutor(Transformer):
2    def __init__(self, market_data):
3        self.data = market_data
4        self.indicators = {}  # Cache computed indicators
5
6    def number(self, items):
7        return float(items[0])
8
9    def price(self, _):
10        return self.data['close']
11
12    def function_call(self, items):
13        func_name = items[0].value
14        period = int(items[1])
15        
16        # Cache key
17        cache_key = f"{func_name}_{period}"
18        if cache_key in self.indicators:
19            return self.indicators[cache_key]
20        
21        # Calculate indicator
22        if func_name == 'RSI':
23            value = self.calculate_rsi(period)
24        elif func_name == 'SMA':
25            value = self.calculate_sma(period)
26        elif func_name == 'EMA':
27            value = self.calculate_ema(period)
28        else:
29            raise ValueError(f"Unknown function: {func_name}")
30        
31        self.indicators[cache_key] = value
32        return value
33
34    def gt(self, items):
35        return items[0] > items[1]
36    
37    def lt(self, items):
38        return items[0] < items[1]
39    
40    def and_cond(self, items):
41        return items[0] and items[1]
42    
43    def or_cond(self, items):
44        return items[0] or items[1]
45
46    def rule(self, items):
47        condition, action = items
48        if condition:
49            return action
50        return "HOLD"
51
52    def buy(self, items):
53        return f"ORDER: BUY {int(items[0])}"
54    
55    def sell(self, items):
56        return f"ORDER: SELL {int(items[0])}"
57
58    def calculate_rsi(self, period):
59        """Calculate RSI indicator."""
60        prices = self.data['history'][-period-1:]
61        gains = [max(prices[i] - prices[i-1], 0) for i in range(1, len(prices))]
62        losses = [max(prices[i-1] - prices[i], 0) for i in range(1, len(prices))]
63        
64        avg_gain = sum(gains) / period
65        avg_loss = sum(losses) / period
66        
67        if avg_loss == 0:
68            return 100
69        
70        rs = avg_gain / avg_loss
71        rsi = 100 - (100 / (1 + rs))
72        return rsi
73    
74    def calculate_sma(self, period):
75        """Calculate Simple Moving Average."""
76        prices = self.data['history'][-period:]
77        return sum(prices) / period
78    
79    def calculate_ema(self, period):
80        """Calculate Exponential Moving Average."""
81        prices = self.data['history'][-period:]
82        multiplier = 2 / (period + 1)
83        ema = prices[0]
84        for price in prices[1:]:
85            ema = (price - ema) * multiplier + ema
86        return ema
87
88### Running It
89
90```python
91import numpy as np
92
93# Simulate market data
94market_state = {
95    'close': 150.0,
96    'history': list(100 + np.cumsum(np.random.randn(100) * 0.5))
97}
98
99strategy_text = "IF RSI(14) > 70 AND PRICE > SMA(50) THEN SELL 100"
100tree = parser.parse(strategy_text)
101
102executor = StrategyExecutor(market_state)
103result = executor.transform(tree)
104print(result)  # Output: ORDER: SELL 100 (if conditions met)
105

1class CppCodeGenerator(Transformer):
2    """Generate C++ code from the AST."""
3    
4    def number(self, items):
5        return str(items[0])
6    
7    def price(self, _):
8        return "ctx.price"
9    
10    def function_call(self, items):
11        func_name = items[0].value.lower()
12        period = items[1]
13        return f"ctx.indicators.{func_name}({period})"
14    
15    def gt(self, items):
16        return f"({items[0]} > {items[1]})"
17    
18    def lt(self, items):
19        return f"({items[0]} < {items[1]})"
20    
21    def and_cond(self, items):
22        return f"({items[0]} && {items[1]})"
23    
24    def or_cond(self, items):
25        return f"({items[0]} || {items[1]})"
26    
27    def buy(self, items):
28        return f"ctx.order_manager.send(Side::BUY, {items[0]})"
29    
30    def sell(self, items):
31        return f"ctx.order_manager.send(Side::SELL, {items[0]})"
32    
33    def rule(self, items):
34        condition = items[0]
35        action = items[1]
36        
37        cpp_code = f"""
38// Generated strategy code
39void execute_strategy(TradingContext& ctx) {{
40    if {condition} {{
41        {action};
42    }}
43}}
44"""
45        return cpp_code
46
47# Generate C++ code
48strategy_text = "IF PRICE > SMA(50) THEN BUY 100"
49tree = parser.parse(strategy_text)
50generator = CppCodeGenerator()
51cpp_code = generator.transform(tree)
52print(cpp_code)
53

1// Generated strategy code
2void execute_strategy(TradingContext& ctx) {
3    if (ctx.price > ctx.indicators.sma(50)) {
4        ctx.order_manager.send(Side::BUY, 100);
5    }
6}
7

1import unittest
2
3class TestTradingDSL(unittest.TestCase):
4    
5    def test_parse_simple_rule(self):
6        """Test parsing a simple rule."""
7        tree = parser.parse("IF PRICE > 100 THEN BUY 50")
8        self.assertIsNotNone(tree)
9    
10    def test_syntax_error(self):
11        """Test syntax error detection."""
12        tree, error = parse_with_errors("IF PRICE > THEN BUY 50")
13        self.assertIsNone(tree)
14        self.assertIn("Syntax error", error)
15    
16    def test_semantic_error_invalid_indicator(self):
17        """Test semantic error for invalid indicator."""
18        with self.assertRaises(SemanticError):
19            tree = parser.parse("IF INVALID(14) > 70 THEN BUY 100")
20            SemanticValidator().transform(tree)
21    
22    def test_type_error(self):
23        """Test type checking."""
24        with self.assertRaises(TypeError):
25            tree = parser.parse("IF RSI(14) THEN BUY 100")  # Missing comparison
26            TypeChecker().transform(tree)
27    
28    def test_execution(self):
29        """Test strategy execution."""
30        market_data = {
31            'close': 150.0,
32            'history': [100 + i * 0.5 for i in range(100)]
33        }
34        
35        tree = parser.parse("IF PRICE > 100 THEN BUY 50")
36        executor = StrategyExecutor(market_data)
37        result = executor.transform(tree)
38        
39        self.assertEqual(result, "ORDER: BUY 50")
40
41if __name__ == '__main__':
42    unittest.main()
43

1import signal
2from contextlib import contextmanager
3
4class TimeoutError(Exception):
5    pass
6
7@contextmanager
8def timeout(seconds):
9    """Context manager for timeout."""
10    def timeout_handler(signum, frame):
11        raise TimeoutError(f"Execution exceeded {seconds} seconds")
12    
13    # Set the signal handler
14    signal.signal(signal.SIGALRM, timeout_handler)
15    signal.alarm(seconds)
16    
17    try:
18        yield
19    finally:
20        signal.alarm(0)
21
22# Use timeout to prevent infinite loops
23try:
24    with timeout(5):
25        # Execute user strategy
26        tree = parser.parse(user_strategy)
27        executor = StrategyExecutor(market_data)
28        result = executor.transform(tree)
29except TimeoutError as e:
30    print(f"Strategy execution timeout: {e}")
31

1class SecureStrategyExecutor(StrategyExecutor):
2    MAX_HISTORY_SIZE = 1000
3    MAX_INDICATOR_PERIOD = 200
4    
5    def __init__(self, market_data):
6        # Limit history size
7        if len(market_data['history']) > self.MAX_HISTORY_SIZE:
8            market_data['history'] = market_data['history'][-self.MAX_HISTORY_SIZE:]
9        
10        super().__init__(market_data)
11    
12    def function_call(self, items):
13        period = int(items[1])
14        
15        # Enforce period limits
16        if period > self.MAX_INDICATOR_PERIOD:
17            raise ValueError(
18                f"Indicator period {period} exceeds maximum {self.MAX_INDICATOR_PERIOD}"
19            )
20        
21        return super().function_call(items)
22