[Proposal] API redesign / Removal of boiler plate code

[MDUYN]

Redesign Strategy Creation API for Rapid Indicator Testing

Problem

The current TradingStrategy class requires ~150 lines to express what is conceptually a simple idea:

"Buy when RSI < 30 AND short EMA crosses above long EMA; sell when RSI ≥ 70 AND short EMA crosses below long EMA."

When a user wants to quickly test a new indicator (e.g. a freshly added PyIndicators function), they must:

1. Register DataSource objects manually (loop over symbols, construct identifier strings)
2. Store indicator column names as instance variables (self.rsi_result_column, etc.)
3. Write a _prepare_indicators helper that calls PyIndicators functions
4. Implement generate_buy_signals / generate_sell_signals — both loop over symbols, look up data by string key, prepare indicators, compose boolean masks
5. Repeat PositionSize, TakeProfitRule, StopLossRule for every symbol even when the config is identical

This friction kills experimentation velocity. Swapping one indicator for another touches 5+ places in the code.

Current API (reference)

class RSIEMACrossoverStrategy(TradingStrategy):
    time_unit = TimeUnit.HOUR
    interval = 2
    symbols = ["BTC"]
    position_sizes = [
        PositionSize(symbol="BTC", percentage_of_portfolio=20.0),
        PositionSize(symbol="ETH", percentage_of_portfolio=20.0),
    ]
    take_profits = [
        TakeProfitRule(symbol="BTC", percentage_threshold=10, trailing=True, sell_percentage=100),
        TakeProfitRule(symbol="ETH", percentage_threshold=10, trailing=True, sell_percentage=100),
    ]
    stop_losses = [
        StopLossRule(symbol="BTC", percentage_threshold=5, trailing=False, sell_percentage=100),
        StopLossRule(symbol="ETH", percentage_threshold=5, trailing=False, sell_percentage=100),
    ]

    def __init__(self, time_unit, interval, market, rsi_time_frame, rsi_period,
                 rsi_overbought_threshold, rsi_oversold_threshold,
                 ema_time_frame, ema_short_period, ema_long_period,
                 ema_cross_lookback_window=10):
        self.rsi_time_frame = rsi_time_frame
        self.rsi_period = rsi_period
        self.rsi_result_column = f"rsi_{self.rsi_period}"
        self.rsi_overbought_threshold = rsi_overbought_threshold
        self.rsi_oversold_threshold = rsi_oversold_threshold
        self.ema_time_frame = ema_time_frame
        self.ema_short_result_column = f"ema_{ema_short_period}"
        self.ema_long_result_column = f"ema_{ema_long_period}"
        self.ema_crossunder_result_column = "ema_crossunder"
        self.ema_crossover_result_column = "ema_crossover"
        self.ema_short_period = ema_short_period
        self.ema_long_period = ema_long_period
        self.ema_cross_lookback_window = ema_cross_lookback_window

        data_sources = []
        for symbol in self.symbols:
            full_symbol = f"{symbol}/EUR"
            data_sources.append(DataSource(
                identifier=f"{symbol}_rsi_data", data_type=DataType.OHLCV,
                time_frame=self.rsi_time_frame, market=market,
                symbol=full_symbol, pandas=True, window_size=800,
            ))
            data_sources.append(DataSource(
                identifier=f"{symbol}_ema_data", data_type=DataType.OHLCV,
                time_frame=self.ema_time_frame, market=market,
                symbol=full_symbol, pandas=True, window_size=800,
            ))
        super().__init__(data_sources=data_sources, time_unit=time_unit, interval=interval)

    def _prepare_indicators(self, rsi_data, ema_data):
        ema_data = ema(ema_data, period=self.ema_short_period, source_column="Close",
                       result_column=self.ema_short_result_column)
        ema_data = ema(ema_data, period=self.ema_long_period, source_column="Close",
                       result_column=self.ema_long_result_column)
        ema_data = crossover(ema_data, first_column=self.ema_short_result_column,
                             second_column=self.ema_long_result_column,
                             result_column=self.ema_crossover_result_column)
        ema_data = crossunder(ema_data, first_column=self.ema_short_result_column,
                              second_column=self.ema_long_result_column,
                              result_column=self.ema_crossunder_result_column)
        rsi_data = rsi(rsi_data, period=self.rsi_period, source_column="Close",
                       result_column=self.rsi_result_column)
        return ema_data, rsi_data

    def generate_buy_signals(self, data):
        signals = {}
        for symbol in self.symbols:
            ema_data, rsi_data = self._prepare_indicators(
                data[f"{symbol}_ema_data"].copy(),
                data[f"{symbol}_rsi_data"].copy(),
            )
            ema_crossover_lookback = ema_data[self.ema_crossover_result_column] \
                .rolling(window=self.ema_cross_lookback_window).max().astype(bool)
            rsi_oversold = rsi_data[self.rsi_result_column] < self.rsi_oversold_threshold
            signals[symbol] = (rsi_oversold & ema_crossover_lookback).fillna(False).astype(bool)
        return signals

    def generate_sell_signals(self, data):
        signals = {}
        for symbol in self.symbols:
            ema_data, rsi_data = self._prepare_indicators(
                data[f"{symbol}_ema_data"].copy(),
                data[f"{symbol}_rsi_data"].copy(),
            )
            ema_crossunder_lookback = ema_data[self.ema_crossunder_result_column] \
                .rolling(window=self.ema_cross_lookback_window).max().astype(bool)
            rsi_overbought = rsi_data[self.rsi_result_column] >= self.rsi_overbought_threshold
            signals[symbol] = (rsi_overbought & ema_crossunder_lookback).fillna(False).astype(bool)
        return signals

~90 lines just for the strategy class, plus the app.add_strategy(...) call with all kwargs repeated. Swapping RSI for e.g. momentum_cycle_sentry means editing 8+ places.

---

Proposal A — Fluent Builder + quick_backtest()

A new high-level API layer that coexists with the current TradingStrategy class (no breaking changes).

A1. quick_backtest() for Notebooks

Zero-class, zero-registration rapid testing:

from investing_algorithm_framework import quick_backtest
from pyindicators import ema, rsi, supertrend

backtest = quick_backtest(
    symbols=["BTC/EUR"],
    market="bitvavo",
    time_frame="4h",
    date_range=("2023-01-01", "2024-06-01"),
    initial_amount=1000,

    # Indicator pipeline — just a callable that receives & returns a DataFrame
    indicators=lambda df: (
        df.pipe(ema, period=12, result_column="ema_short")
          .pipe(ema, period=26, result_column="ema_long")
          .pipe(rsi, period=14, result_column="rsi")
          .pipe(supertrend, atr_length=10, factor=3.0)
    ),

    # Signals — simple callables returning bool Series
    buy_when=lambda df: (df["rsi"] < 30) & (df["supertrend_signal"] == 1),
    sell_when=lambda df: (df["rsi"] > 70) | (df["supertrend_signal"] == -1),

    position_size=0.2,        # 20% of portfolio per trade
    take_profit=0.10,         # 10% trailing TP
    stop_loss=0.05,           # 5% SL
)

BacktestReport(backtest).show()  # Opens the backtest report in browser

~20 lines. Swap supertrend for momentum_cycle_sentry — change 2 lines.

A2. Fluent Strategy Builder

For strategies that need to be registered with the app (live trading / scheduled execution):

from investing_algorithm_framework import Strategy, Signal

strategy = (
    Strategy("RSI EMA Crossover")
    .symbols(["BTC", "ETH"])
    .market("bitvavo")
    .time_frame("2h")
    .interval(2, TimeUnit.HOUR)
    .position_size(percentage=20)               # uniform for all symbols
    .take_profit(percentage=10, trailing=True)
    .stop_loss(percentage=5)

    # Declare indicators — data sources auto-inferred
    .indicator(ema, period=12, source_column="Close", name="ema_short")
    .indicator(ema, period=26, source_column="Close", name="ema_long")
    .indicator(rsi, period=14, source_column="Close", name="rsi")

    # Composable signal conditions
    .buy_when(
        Signal("rsi", "<", 30)
        & Signal.crossover("ema_short", "ema_long", lookback=10)
    )
    .sell_when(
        Signal("rsi", ">=", 70)
        & Signal.crossunder("ema_short", "ema_long", lookback=10)
    )
)

app = create_app()
app.add_strategy(strategy)

backtest = app.run_backtest(....)
BacktestReport(backtest).show()  # Opens the backtest report in browser

A3. Signal Composables

Reusable signal building blocks:

from investing_algorithm_framework import Signal

# These map directly to PyIndicators *_signal() output columns
supertrend_bull = Signal("supertrend_signal", "==", 1)
rsi_oversold    = Signal("rsi", "<", 30)
mcs_bullish     = Signal("mcs_signal", "==", 1)

# Compose with &, |, ~
entry = supertrend_bull & rsi_oversold & mcs_bullish
exit  = ~supertrend_bull | Signal("rsi", ">", 70)

Pros of Proposal A:

• Most concise API possible
• Great for notebooks and rapid prototyping
• Signal objects are reusable across strategies
• Familiar fluent / builder pattern

Cons of Proposal A:

• Builder pattern can feel "magic" — harder to debug when something goes wrong
• Less explicit about data flow
• Another abstraction layer to maintain alongside TradingStrategy

---

Proposal B — Simplified TradingStrategy with Smart Defaults (No Builder Pattern)

Instead of adding a new abstraction, improve the existing TradingStrategy class with three changes:

B1. Indicator Descriptors — Auto-Register Data Sources

Declare indicators as class attributes. The metaclass / __init_subclass__ hook auto-creates the DataSource objects:

from investing_algorithm_framework import TradingStrategy, Indicator, PositionSize
from pyindicators import ema, rsi, crossover, crossunder

class RSIEMACross(TradingStrategy):
    time_unit = TimeUnit.HOUR
    interval = 2
    market = "bitvavo"
    time_frame = "2h"
    symbols = ["BTC", "ETH"]
    window_size = 800

    # Risk rules — apply uniformly unless overridden per symbol
    position_size = PositionSize(percentage_of_portfolio=20)
    take_profit = TakeProfitRule(percentage_threshold=10, trailing=True)
    stop_loss = StopLossRule(percentage_threshold=5)

    # Indicator descriptors — auto-register data sources
    ema_short = Indicator(ema, period=12, source_column="Close")
    ema_long  = Indicator(ema, period=26, source_column="Close")
    rsi_val   = Indicator(rsi, period=14, source_column="Close")

    def prepare(self, df):
        """Apply all indicators + derived columns. Called once per symbol per tick."""
        df = self.ema_short.apply(df)
        df = self.ema_long.apply(df)
        df = crossover(df, first_column=self.ema_short.column, 
                       second_column=self.ema_long.column,
                       result_column="ema_crossover")
        df = crossunder(df, first_column=self.ema_short.column,
                        second_column=self.ema_long.column,
                        result_column="ema_crossunder")
        df = self.rsi_val.apply(df)
        return df
 
   def generate_buy_signals(self, data):
        signals = {}
        for symbol in self.symbols:
            ema_data, rsi_data = self._prepare_indicators(
                data[f"{symbol}_ema_data"].copy(),
                data[f"{symbol}_rsi_data"].copy(),
            )
            ema_crossover_lookback = ema_data[self.ema_crossover_result_column] \
                .rolling(window=self.ema_cross_lookback_window).max().astype(bool)
            rsi_oversold = rsi_data[self.rsi_result_column] < self.rsi_oversold_threshold
            signals[symbol] = (rsi_oversold & ema_crossover_lookback).fillna(False).astype(bool)
        return signals

   

   # Is used by def generate_buy_signals(self, data):
   def buy_signal(self, df) -> pd.Series:
        """Return a boolean Series. Framework handles the per-symbol loop."""
        ema_cross_recent = df["ema_crossover"].rolling(window=10).max().astype(bool)
        return (df[self.rsi_val.column] < 30) & ema_cross_recent
   
# Is used by def generate_sell_signals(self, data):
    def sell_signal(self, df) -> pd.Series:
        ema_cross_recent = df["ema_crossunder"].rolling(window=10).max().astype(bool)
        return (df[self.rsi_val.column] >= 70) & ema_cross_recent

What changed vs. current API:

Aspect	Current	Proposed B
Data sources	Manual loop + string identifiers	Auto-inferred from Indicator descriptors
Indicator columns	Instance variables (self.rsi_result_column)	self.rsi_val.column (auto-named)
Risk rules	Repeated per symbol in lists	Single instance, auto-applied to all symbols
Signal methods	generate_buy_signals returns Dict[str, Series] with manual symbol loop	buy_condition returns single Series, framework handles loop
Indicator application	Manual in _prepare_indicators	prepare(df) with indicator.apply(df) helpers
Lines of code	~90	~35

B2. Uniform Risk Management

PositionSize, TakeProfitRule, StopLossRule accept a single instance that applies to all symbols:

# Current — must repeat for every symbol
position_sizes = [
    PositionSize(symbol="BTC", percentage_of_portfolio=20.0),
    PositionSize(symbol="ETH", percentage_of_portfolio=20.0),
]

# Proposed — uniform (symbol is optional, defaults to all)
position_size = PositionSize(percentage_of_portfolio=20)

# Per-symbol override still supported via dict
position_sizes = {
    "BTC": PositionSize(percentage_of_portfolio=30),
    "ETH": PositionSize(percentage_of_portfolio=10),
}

B3. quick_backtest() as a Standalone Function

Even without the builder, a functional quick_backtest() can coexist:

from investing_algorithm_framework import quick_backtest
from pyindicators import momentum_cycle_sentry, momentum_cycle_sentry_signal

def my_indicators(df):
    df = momentum_cycle_sentry(df, length=20, smoothing=5)
    df = momentum_cycle_sentry_signal(df)
    return df

results = quick_backtest(
    symbols=["BTC/EUR"],
    market="bitvavo",
    time_frame="4h",
    date_range=("2023-01-01", "2024-06-01"),
    initial_amount=1000,
    indicators=my_indicators,
    buy_when=lambda df: df["mcs_signal"] == 1,
    sell_when=lambda df: df["mcs_signal"] == -1,
    position_size=0.2,
    take_profit=0.10,
    stop_loss=0.05,
)
results.show()

Pros of Proposal B:

• No new abstraction layer — extends the existing TradingStrategy class users already know
• Fully explicit data flow (no hidden magic)
• Indicator descriptor is a thin wrapper, easy to understand and debug
• Backward compatible — old strategies keep working, new ones benefit from less boilerplate
• quick_backtest() function works independently for notebook prototyping

Cons of Proposal B:

• Still requires a class for anything beyond quick_backtest()
• Slightly more verbose than Proposal A's builder

---

Indicator Descriptor — Implementation Sketch

Both proposals use the Indicator descriptor. Here's the core idea:

class Indicator:
    """Wraps a PyIndicators function for declarative use in strategies."""

    def __init__(self, func, *, name=None, **kwargs):
        self.func = func
        self.name = name  # set by __set_name__ if not provided
        self.kwargs = kwargs
        self.column = None  # resolved after apply()

    def __set_name__(self, owner, name):
        if self.name is None:
            self.name = name
        # Auto-determine result column name
        self.column = self.kwargs.get("result_column", self.name)
        self.kwargs.setdefault("result_column", self.column)

    def apply(self, df):
        """Apply the indicator function to the DataFrame."""
        return self.func(df, **self.kwargs)

    def __repr__(self):
        return f"Indicator({self.func.__name__}, column={self.column!r})"

---

Comparison at a Glance

Feature	Current	Proposal A (Builder)	Proposal B (Improved Class)
Lines for RSI+EMA strategy	~90	~20	~35
New abstraction layers	—	Strategy, Signal, quick_backtest	Indicator, quick_backtest
Data source registration	Manual	Auto	Auto
Risk rules per symbol	Explicit list	Uniform default	Uniform default
Signal composition	Imperative code	Declarative Signal objects	Imperative (simpler methods)
Notebook prototyping	Not supported	quick_backtest()	quick_backtest()
Backward compatible	—	Yes (additive)	Yes (additive)
Learning curve for new users	High	Low (but magic)	Medium
Debugging transparency	Good	Lower (builder hides flow)	Good

---

Recommendation

Start with Proposal B — it has the highest value-to-risk ratio:

1. quick_backtest() function — immediate impact for notebook prototyping (P0)
2. Indicator descriptor — eliminates boilerplate without hiding logic (P1)
3. Uniform risk rules — small change, big reduction in repetition (P1)

Proposal A's Strategy builder and Signal composables can be added later as a convenience layer on top of Proposal B, once the foundation is solid.

---

Suggested Implementation Order

• Implement Indicator descriptor class
• Add __init_subclass__ hook to TradingStrategy for auto data source registration
• Support single PositionSize / TakeProfitRule / StopLossRule (uniform for all symbols)
• Add prepare(df) / buy_condition(df) / sell_condition(df) method pattern
• Implement quick_backtest() standalone function
• Add notebook examples demonstrating rapid indicator testing
• (Future) Signal composable objects
• (Future) Fluent Strategy builder