
Quick jump:
- 1 Embracing SOLID Principles and PEP 8
- 2 Defining a Clear Strategy Interface and Factory
- 3 Structuring Telegram Handlers for Clarity
- 4 Enhancing Strategy Selection with Inline Keyboards
- 5 Improving Notifications: Handling Blocked‐Bot Scenarios
- 6 Bringing It All Together
- 7 EODHDAlerts in Action
- 8 Reflections and Next Steps
When I first architected and developed the Telegram trading bot, I approached it as a practical prototype: fetch OHLC data, apply a simple moving average (SMA) crossover, and send a concise “Buy” or “Sell” notifications via Telegram. At that stage, rapid iteration and an MVP was my priority. However, as I began integrating additional indicators—RSI, MACD, Bollinger Bands, and so on—the code started to exhibit signs of strain. Each new strategy required duplicating portions of signal‐generation logic, renaming rolling averages, and ensuring consistency across multiple modules. Although the initial implementation was functional, it soon became evident that a cleaner, more maintainable architecture was essential. In short, I needed to elevate the bot from a proof‐of‐concept to something that adhered to established development practices.
Embracing SOLID Principles and PEP 8
Rather than simply patching over incremental complexity, I decided on a concerted refactor guided by two pillars: PEP8 conventions and the SOLID design principles. My objectives were:
- Single Responsibility: Ensure each module or class handles one clear aspect of functionality—no more tangled responsibilities in a single file.
- Open/Closed: Allow the code to be extended with new strategies without modifying existing, well‐tested logic.
- Liskov Substitution: Guarantee that any strategy implementation could be swapped in for another without altering downstream behaviour.
- Interface Segregation: Define minimal, specific interfaces that reflect an individual component’s needs rather than a monolithic interface that tries to do everything.
- Dependency Inversion: Depend on abstractions (strategy interfaces), not concrete implementations (specific moving averages).
Complementing these principles, I committed to consistent naming and formatting—snake_case for functions and variables, PascalCase for classes, clear docstrings, and proper line lengths. This ensured that every contributor (including my future self) could navigate the codebase without wrestling with idiosyncratic style choices.
Defining a Clear Strategy Interface and Factory
The first step was to isolate all indicator logic into a single module, strategy.py. At its core lies an abstract base class:
from abc import ABC, abstractmethod
import pandas as pd
class Strategy(ABC):
@abstractmethod
def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
By specifying exactly one method, I guarantee that any subclass—whether it computes an SMA crossover, an RSI threshold, or something more exotic—will honour the same contract. That means the remainder of the system can treat them interchangeably.
Using this interface, I implemented five concrete strategies, each in its own well‐named class:
- SMA Crossover (SmaCrossoverStrategy)
- EMA Crossover (EmaCrossoverStrategy)
- RSI‐Based (RsiStrategy)
- MACD (MacdStrategy)
- Bollinger Bands Breakout (BollingerBandsStrategy)
Rather than re‐implementing every rolling window or exponential average by hand, and risk subtle off‐by‐one errors or NaN handling issues, I leveraged the well‐maintained pandas_ta library. For example, the EMA crossover class now looks like this:
import pandas_ta as ta
class EmaCrossoverStrategy(Strategy):
def __init__(self, short_span: int = 12, long_span: int = 26):
self.short_span = short_span
self.long_span = long_span
def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
df = data.copy()
df[f'ema_{self.short_span}'] = ta.ema(df['close'], length=self.short_span)
df[f'ema_{self.long_span}'] = ta.ema(df['close'], length=self.long_span)
df['signal'] = 0
df.loc[df[f'ema_{self.short_span}'] > df[f'ema_{self.long_span}'], 'signal'] = 1
df.loc[df[f'ema_{self.short_span}'] < df[f'ema_{self.long_span}'], 'signal'] = -1
df['position'] = df['signal'].diff().fillna(0)
return df
A single line of pandas_ta replaces what used to be a multi‐line rolling‐average implementation, greatly reducing maintenance overhead. I applied the same pattern to RSI, MACD and Bollinger Bands, keeping each class limited to its own calculation and threshold logic.
To manage instantiation by name, I introduced a simple StrategyFactory:
class StrategyFactory:
_strategies = {
'sma': SmaCrossoverStrategy,
'ema': EmaCrossoverStrategy,
'rsi': RsiStrategy,
'macd': MacdStrategy,
'bbands': BollingerBandsStrategy,
}
@classmethod
def list_strategies(cls) -> list[str]:
return list(cls._strategies.keys())
@classmethod
def create_strategy(cls, name: str, **kwargs) -> Strategy:
name = name.lower()
if name not in cls._strategies:
raise ValueError(f'Unknown strategy: {name}')
return cls._strategies[name](**kwargs)
With this factory, adding a sixth strategy later is simply a matter of implementing a new subclass and registering it in _strategies. The rest of the code – handlers, notifier, job scheduling remains untouched, fulfilling the Open/Closed principle.
Structuring Telegram Handlers for Clarity
Next, I focused on handlers.py, ensuring each function has a single, obvious responsibility:
- start: issue a welcome message upon /start.
- set_symbol: parse /set_symbol <SYMBOL>, store it in a module‐level variable (uppercase by convention), and confirm.
- set_interval: handle /set_interval <INTERVAL> similarly.
- get_price: fetch only the most recent quote when you need a quick check.
- list_strategies: display all registered strategy keys.
- set_strategy: allow strategy selection—detailed below.
- current_strategy: reply with which strategy is active.
- analyse_market: the scheduled job that fetches OHLC data, runs the active strategy, and sends “Buy” or “Sell” as needed.
def get_price(update: Update, context: CallbackContext) -> None:
fetcher = DataFetcher(symbol, interval)
price = fetcher.fetch_price()
if price is not None:
update.message.reply_text(f'Current price of {symbol}: {price}')
else:
update.message.reply_text('Failed to fetch data.')
Notice that none of these handler functions attempt to compute signals or manage strategy internals. They simply orchestrate inputs and outputs, deferring the core logic to the StrategyFactory and notifier.
Enhancing Strategy Selection with Inline Keyboards
Originally, choosing a strategy required an exact command, for example:
/set_strategy ema
That step was functional, but I wanted a more intuitive developer experience—especially if I didn’t recall every available key by heart. Therefore, I extended the /set_strategy handler so that if no argument is provided, the bot sends an inline keyboard:
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
def set_strategy(update: Update, context: CallbackContext) -> None:
global current_strategy_name, strategy_params
names = StrategyFactory.list_strategies()
if context.args:
name = context.args[0].lower()
if name in names:
current_strategy_name = name
strategy_params = {}
update.message.reply_text(f'Strategy set to "{name}".')
else:
update.message.reply_text(f'Unknown strategy "{name}".')
return
buttons = [
InlineKeyboardButton(text=nm.upper(), callback_data=f'setstrat:{nm}')
for nm in names
]
# Arrange buttons in two columns
rows = [buttons[i:i+2] for i in range(0, len(buttons), 2)]
reply_markup = InlineKeyboardMarkup(rows)
update.message.reply_text('Please choose a strategy:', reply_markup=reply_markup)
That command now displays a neatly laid‐out grid of buttons—“SMA”, “EMA”, “RSI”, and so on. When the user taps a button, Telegram sends a callback query with data set to something like setstrat:ema. I handle these callbacks with:
def strategy_button(update: Update, context: CallbackContext) -> None:
global current_strategy_name, strategy_params
query = update.callback_query
query.answer() # Dismisses the “Loading...” spinner
data = query.data or ''
_, chosen = data.split(':', 1)
if chosen in StrategyFactory.list_strategies():
current_strategy_name = chosen
strategy_params = {}
query.edit_message_text(f'Strategy set to "{chosen}".')
else:
query.edit_message_text(f'Unknown strategy "{chosen}".')
Crucially, query.answer() must be called to immediately clear the loading indicator. Otherwise, users might tap a button only to see an indefinite spinner. By editing the original message with a confirmation, I provide immediate feedback that the selection was successful. In main.py, the order of handler registration is:
dp.add_handler(CommandHandler("set_strategy", set_strategy))
dp.add_handler(CallbackQueryHandler(strategy_button, pattern=r"^setstrat:"))
Ensuring the CallbackQueryHandler is registered after the command handlers guarantees that button presses will be caught. With this setup, the user can simply type:
/set_strategy
and tap “EMA”, rather than remember the exact “ema” key.
Improving Notifications: Handling Blocked‐Bot Scenarios
In production, one of the most common pitfalls is uncaught exceptions due to a user blocking the bot. At one point, my scheduler would crash with:
telegram.error.Unauthorized: Forbidden: bot was blocked by the user
Because I sent messages directly in the job callback, any blocked‐bot error terminated the entire process. To harden against this, I refactored all message‐sending logic into a dedicated TelegramNotifier:
import logging
from telegram.error import Unauthorized, BadRequest
from telegram import Bot
class TelegramNotifier:
def __init__(self, bot_token: str, default_chat_id: int = None):
self._bot = Bot(bot_token)
self._default_chat_id = default_chat_id
def send_message(self, text: str, chat_id: int = None) -> None:
target = chat_id or self._default_chat_id
if not target:
logging.error("No chat_id provided; cannot send message.")
return
try:
self._bot.send_message(chat_id=target, text=text)
except Unauthorized:
logging.warning(f"Failed to send to {target}: bot was blocked.")
# TODO: remove this chat_id from scheduled jobs or notify an admin
except BadRequest as e:
logging.error(f"BadRequest sending to {target}: {e}")
except Exception as e:
logging.exception(f"Unexpected error sending to {target}: {e}")
Now, even if a user blocks the bot, the Unauthorized exception is caught and logged, and the scheduler continues without interruption. If I maintain a list of active chat IDs or schedule jobs per chat, I can remove blocked chats from that list at this point.
Bringing It All Together
With these refinements, the bot’s architecture is neatly partitioned:
- strategy.py
- Defines a single Strategy interface.
- Implements five concrete strategies (SMA, EMA, RSI, MACD, Bollinger Bands), each using pandas_ta.
- Provides a StrategyFactory to list and instantiate strategies by name.
- Defines a single Strategy interface.
- handlers.py
- Houses Telegram command handlers (/start, /set_symbol, /set_interval, /get_price, /list_strategies, /set_strategy, /current_strategy).
- Includes inline‐keyboard logic for /set_strategy when no argument is provided.
- Contains the scheduled job callback analyse_market, which fetches OHLC data, invokes the active strategy via StrategyFactory, and uses TelegramNotifier to dispatch buy/sell alerts.
- Houses Telegram command handlers (/start, /set_symbol, /set_interval, /get_price, /list_strategies, /set_strategy, /current_strategy).
- main.py
- Registers all CommandHandler instances.
- Registers a CallbackQueryHandler for inline buttons with pattern=r”^setstrat:”.
- Schedules analyse_market on a repeating interval (e.g., 60 seconds).
- Registers all CommandHandler instances.
By adhering to the Single Responsibility principle, each component focuses on one aspect E.g., strategy logic, Telegram interaction, or scheduling, not muddling concerns together. The Open/Closed principle is satisfied because adding a new strategy never forces me to alter existing handler or notifier code; I simply implement a new subclass of Strategy and register it in _strategies. Types remain substitutable, honouring Liskov Substitution, and the code depends on abstractions rather than concrete classes, in line with Dependency Inversion.
EODHDAlerts in Action

Reflections and Next Steps
Working through this refactor reinforced how valuable a clear architecture can be, even for a relatively compact project. By isolating strategy logic into an interface and factory, I can now introduce complex, data‐driven models or custom indicators without touching the Telegram‐specific code. The inline keyboards in /set_strategy create a smoother experience, no more guesswork over exact strategy names and wrapping send_message in robust error handling ensures that blocked‐bot scenarios don’t bring the entire scheduler down.
Moving forward, a few enhancements are on my radar:
- Per‐chat persistence: store each chat’s chosen symbol, interval and strategy in a lightweight database or JSON file so that settings persist across bot restarts.
- Parameterised commands: extend /set_strategy to accept parameters (for example, /set_strategy sma 10 30 to choose custom window lengths).
- Back‐testing integration: reuse the same Strategy classes to run historical simulations on OHLC data and generate performance reports.
- Additional strategies: including machine‐learning classifiers or composite oscillators, simply by subclassing Strategy and registering.
Ultimately, this refactor has allowed me to maintain both flexibility and stability. I can deploy new ideas rapidly, while ensuring the codebase remains understandable and maintainable. For anyone building a Telegram trading assistant or any system that needs to swap algorithms on the fly, this approach demonstrates how even a modest project benefits from well‐established design principles.