Skip to content

contango.strategy.rule_based.rule_based_strategy

contango.strategy.rule_based.rule_based_strategy

RuleBasedStrategy 

RuleBasedStrategy(streams: Sequence[Stream[Any]], rules: Sequence[Rule[TradingContext, Intent | None]], position_sizer: PositionSizer[TradingContext])

Bases: Strategy

Allows rules to be executed upon and intents to be executed.

Initializes RuleBasedStrategy.

Parameters:

Name Type Description Default
streams Sequence[Stream[Any]]

A list of streams to be updated with context every market data event.

 required
rules Sequence[Rule[TradingContext, Intent | None]]

A sequence of rules, where each rule returns their intent type or None (executed in order).

 required
position_sizer PositionSizer[TradingContext]

The sizing rule for all trades made.

 required
Source code in src\contango\strategy\rule_based\rule_based_strategy.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(
    self,
    streams: Sequence[Stream[Any]],
    rules: Sequence[Rule[TradingContext, Intent | None]],
    position_sizer: PositionSizer[TradingContext],
) -> None:
    """
    Initializes `RuleBasedStrategy`.

    Args:
        streams: A list of streams to be updated with context every market data event.
        rules: A sequence of rules, where each rule returns their intent type or None (executed in order).
        position_sizer: The sizing rule for all trades made.
    """
    self._streams = streams
    self._rules = rules
    self._position_sizer = position_sizer
    self._last_event: MarketDataEvent | None = None

on_end 

on_end() -> None

Closes any open long position at the final market event.

Source code in src\contango\strategy\rule_based\rule_based_strategy.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
@final
def on_end(self) -> None:
    """
    Closes any open long position at the final market event.
    """
    final_event = self._last_event

    if final_event is None:
        raise ValueError("The final market data event was never set.")

    size = self.portfolio_snapshot.position
    if size > 0:
        self.order_api.submit_order(final_event, final_event.symbol, -size)

on_market_event 

on_market_event(event: MarketDataEvent) -> None

Updates all streams, executes rules, and executes intents every event.

Source code in src\contango\strategy\rule_based\rule_based_strategy.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@final
def on_market_event(self, event: MarketDataEvent) -> None:
    """
    Updates all streams, executes rules, and executes intents every event.
    """
    self._last_event = event

    for stream in self._streams:
        stream.update(event)
    self._position_sizer.update(event)

    context = TradingContext(
        event=event,
        portfolio=self.portfolio_snapshot,
    )

    for rule in self._rules:
        intent = rule.evaluate(context)
        if intent is not None:
            self._translate(intent, context)