FTX Algotrading is a simple algorithmic trading strategy runner for FTX.
The project aims to provide an environment and tools for developing automatic trading strategies. It comes with a simple framework that can be used to automatically retrieve OHLC data within several timeframes, compute technical indicators and setup managed position on FTX exchange.
This project comes with some built-in strategies. Don't hesitate to take a look at it to better understand how the project works.
Open config/ folder and fill the private/ folder with a new file called ftx_config.py with your ftx api key
info. A template is given in this directory with some basic config info about ftx exchange and how to structure the
configuration file.
ws_endpoint = "wss://ftx.com/ws/"
rest_endpoint = "https://ftx.com/api/"
api = {
"key": "YOUR API KEY HERE",
"secret": "YOUR API SECRET HERE",
"sub_account": "YOUR SUB ACCOUNT NAME HERE"
}config/application_config.py allows configuring what strategy to run. It also permits setting up log path and level.
from strategies.demo_strategy.demo_strategy import DemoStrategy
strategy = DemoStrategy()
log = {
"level": "info",
"path": "logs"
}Strategies are stored into strategies/ folder.
Let's create a demo_strategy to illustrate this documentation:
- Create a
strategies/demo_strategyfolder - Create the strategy file
strategies/demo_strategy/demo_strategy.py
All strategies must extend the Strategy class. Here is a minimum working strategy code:
import logging
import time
from core.strategy.strategy import Strategy
class DemoStrategy(Strategy):
"""The demo strategy"""
def __init__(self):
"""The demo strategy constructor"""
logging.info("DemoStrategy run strategy")
super(DemoStrategy, self).__init__()
def before_loop(self) -> None:
"""Called before each loop"""
logging.info("DemoStrategy before_loop")
pass
def loop(self) -> None:
"""The strategy core loop method"""
logging.info("DemoStrategy loop")
pass
def after_loop(self) -> None:
"""Called after each loop"""
logging.info("DemoStrategy after_loop")
time.sleep(10) # Sleep 10 sec
def cleanup(self) -> None:
"""Clean strategy execution"""
logging.info("DemoStrategy cleanup")A strategy is structured around 5 key methods. It works this way:
__init__ is called first as it's the class constructor, then the strategy runner will call indefinitely before_loop,
loop and after_loop. When a blocking exception is raised or when the program is getting killed, cleanup is called.
__init__contains your strategy initialization logics. You can set up FTX data acquisition, initialize vars or whateverbefore_loopis called before each strategy loop. You can use it to ensure all is ready for the loop core methodloopis the strategy loop core method where you will most likely put the core logics of your strategy. Market data checks, wallet balance recovery, decide to open or not a position and drive opened position for example are logics to be put into this methodafter_loopis called after each strategy loop. You can use it to clean anything you used in the loop core method and to make your strategy sleep a bit before the next loopcleanupcontains your strategy cleanup logics. You can delete file, close position or whatever
In order to launch stock data acquisition, you will need a
FtxRestApi
instance, a trading pair (BTC-PERP for example), and the different timeframes you want to retrieve data on.
Timeframes are expressed in seconds and supported values are: [15, 60, 300, 900, 3600, 14400, 86400]
You can launch as many data acquisitions as you want on several coins as long as it doesn't exceed FTX API rate limits
Let's add some logic to the demo_strategy developed in the Create a strategy section to launch
background data acquisition for BTC-PERP on 15 sec and 60 sec timeframes:
In the __init__ method, launch data acquisition:
from core.ftx.rest.ftx_rest_api import FtxRestApi
from core.stock.crypto_pair_manager import CryptoPairManager
def __init__(self):
"""The demo strategy constructor"""
logging.info("DemoStrategy run strategy")
super(DemoStrategy, self).__init__()
self.ftx_rest_api: FtxRestApi = FtxRestApi()
self.btc_pair_manager: CryptoPairManager = CryptoPairManager("BTC-PERP", self.ftx_rest_api)
self.btc_pair_manager.add_time_frame(15)
self.btc_pair_manager.add_time_frame(60)
self.btc_pair_manager.start_all_time_frame_acq()In the cleanup method, stop data acquisition:
def cleanup(self) -> None:
"""Clean strategy execution"""
logging.info("DemoStrategy cleanup")
self.btc_pair_manager.stop_all_time_frame_acq()Launch the bot and see the data acquisition being done in front of your eyes.
⚠️ After starting the acquisition, the last MAX_ITEM_IN_DATA_SET will be retrieved (see in StockDataManager). Older data points will be removed from the system as the acquisition continues. If you want to retrieve more than this amount of data, store it in your strategy or update the global MAX_ITEM_IN_DATA_SET value.
Acquired data are OHLCV (Open, High, Low, Close, Volume) points used to represent candles in candlestick charts. The Candle class provide OHLCV values for each point in addition to some helpers that identify candle patterns such as hammer, hanging man, inverted hammer or shooting star.
Let's continue to work on the demo_strategy developed in the Create a strategy section after we
succeeded to Launch stock data acquisition. We will now display some information about
the last data points.
In the loop method, read last data point and display last 3 data points average volume:
from core.stock.stock_data_manager import StockDataManager
from core.models.candle import Candle
def loop(self) -> None:
"""The strategy core loop method"""
logging.info("DemoStrategy loop")
stock_data_manager: StockDataManager = self.btc_pair_manager.get_time_frame(15).stock_data_manager
# Display last candle info
if len(stock_data_manager.stock_data_list) > 1:
last_candle: Candle = stock_data_manager.stock_data_list[-1]
logging.info(f"Last candle open price: {last_candle.open_price}")
logging.info(f"Last candle high price: {last_candle.high_price}")
logging.info(f"Last candle low price: {last_candle.low_price}")
logging.info(f"Last candle close price: {last_candle.close_price}")
logging.info(f"Last candle volume: {last_candle.volume}")
logging.info(f"Last candle is a hammer or a hanging man: {last_candle.is_hammer_or_hanging_man()}")
# Display last 3 candles average volume
if len(stock_data_manager.stock_data_list) > 3:
last_3_candle_volumes = sum([d.volume for d in stock_data_manager.stock_data_list[-3:]])
logging.info(f"Last 3 candles average volume: {last_3_candle_volumes / 3}")Technical indicators are automatically computed after each data acquisition (See Disable / enable automatically computed technical indicators if needed). We use stockstats library that supply a wrapper StockDataFrame for pandas.DataFrame with inline stock statistics/indicators support.
There are plenty of indicators supported, the documentation is clear, and the library is quite easy to handle. Here is a
basic example of how to update the demo_strategy to display the last RSI (Relative Strength Index) values:
In the loop method, add this code to read RSI stockstats values:
import stockstats
stock_data_manager: StockDataManager = self.btc_pair_manager.get_time_frame(15).stock_data_manager
# Use stockstats to display RSI_14 indicator values
indicators_dataframe: stockstats.StockDataFrame = stock_data_manager.stock_indicators
if indicators_dataframe is not None:
logging.info(indicators_dataframe['rsi'])The FtxRestApi class allows interfacing to the FTX Api directly. For full detail on what you can do with FTX REST Api, see FTX REST Api documentation.
First step is to create an FtxRestApi instance:
from core.ftx.rest.ftx_rest_api import FtxRestApi
ftx_rest_api: FtxRestApi = FtxRestApi()Then you can use the FtxRestApi instance to retrieve information from FTX Api. See some examples:
from tools.utils import format_wallet_raw_data
# Get last market data
response = ftx_rest_api.get(f"markets/BTC-PERP")
logging.info(f"FTX API response: {str(response)}")
# Get open orders on a given market
response = ftx_rest_api.get("orders", {"market": "BTC-PERP"})
logging.info(f"FTX API response: {str(response)}")
# Get your account wallet balances
response = ftx_rest_api.get("wallet/balances")
logging.info(f"FTX API response: {str(response)}")
# Display USD wallet info
wallets = [format_wallet_raw_data(wallet) for wallet in response if
wallet["coin"] == 'USD']
logging.info(f"FTX USD Wallet: {str(wallets)}")
# ...There are some other FTX use examples in the existing strategies, feel free to have a look at them or to dive into FTX documentation.
PositionDriver class allows running a position with automated management. It allows creating simple position opening setup with trigger orders for taking profit or stopping losses.
Let's open the following position:
- Market buy 0.001
BTC-PERP - Stop loss 100% of the position size at -1% of the current price
- Take profit 50% of the position size at +3% of the current price
- Take profit the other 50% of the position size at +6% of the current price
- Add a trailing stop for 100% of the position size at -2% of the current price
from core.models.opening_config_dict import OpeningConfigDict
from core.enums.order_type_enum import OrderTypeEnum
from core.models.position_config_dict import PositionConfigDict
from core.models.trigger_order_config_dict import TriggerOrderConfigDict
from core.enums.side_enum import SideEnum
from core.enums.trigger_order_type_enum import TriggerOrderTypeEnum
from core.models.market_data_dict import MarketDataDict
from tools.utils import format_market_raw_data
from core.trading.position_driver import PositionDriver
from typing import List
# Retrieve market data
logging.info("Retrieving market price")
response = ftx_rest_api.get(f"markets/BTC-PERP")
market_data: MarketDataDict = format_market_raw_data(response)
openings: List[OpeningConfigDict] = [{
"price": None,
"size": 0.001,
"type": OrderTypeEnum.MARKET
}]
sl: TriggerOrderConfigDict = {
"size": 0.001,
"type": TriggerOrderTypeEnum.STOP,
"reduce_only": True,
"trigger_price": market_data["ask"] - market_data["ask"] * 1 / 100,
"order_price": None,
"trail_value": None
}
tp1: TriggerOrderConfigDict = {
"size": 0.0005,
"type": TriggerOrderTypeEnum.TAKE_PROFIT,
"reduce_only": True,
"trigger_price": market_data["ask"] + market_data["ask"] * 3 / 100,
"order_price": None,
"trail_value": None
}
tp2: TriggerOrderConfigDict = {
"size": 0.0005,
"type": TriggerOrderTypeEnum.TAKE_PROFIT,
"reduce_only": True,
"trigger_price": market_data["ask"] + market_data["ask"] * 6 / 100,
"order_price": None,
"trail_value": None
}
trailing_stop: TriggerOrderConfigDict = {
"size": 0.001,
"type": TriggerOrderTypeEnum.TRAILING_STOP,
"reduce_only": True,
"trigger_price": None,
"order_price": None,
"trail_value": market_data["ask"] * -2 / 100
}
position_config: PositionConfigDict = {
"openings": openings,
"trigger_orders": [sl, tp1, tp2, trailing_stop],
"max_open_duration": 60 * 60 * 4 # 4 hours
}
position_driver: PositionDriver = PositionDriver(ftx_rest_api)
position_driver.open_position("DOGE-PERP", SideEnum.BUY, position_config)Once a position is opened, the PositionDriver will check every 10 secs your existing position on FTX and update its internal state automatically if the position is closed / liquidated. As order Api calls have a high impact in Ftx Api call rate limit, a 10 sec check can be a too high frequency. You can update this value when using the PositionDriver constructor:
# Set PositionDriver check frequency to 120 sec
position_driver: PositionDriver = PositionDriver(ftx_rest_api, 120)
⚠️ When opening a LIMIT order, the order may not be filled immediately. As a consequence, the trigger_orders will be created all at once after the first PositionDriver check succeeding a successful opening. For finer / more specific management, please use FTX Api directly.
You can read the PositionDriver current state using the following code:
from core.enums.position_state_enum import PositionStateEnum
# Opened / not opened
position_state: PositionStateEnum = position_driver.position_stateWhen you have a lot of data to acquire, it can be better to not display logs since there are not necessarily relevant.
Default value for this configuration is True.
from core.stock.time_frame_manager import TimeFrameManager
# Deactivate stock data log for readability purposes
TimeFrameManager.log_received_stock_data = FalseWhen a timeframe is running and acquiring data, the default behaviour is to compute and refresh a bunch of technical
indicators on each candle retrieved. If this is not needed, this behaviour can be disabled for performance purposes.
When adding a timeframe to a given
CryptoPairManager,
set auto_compute_indicators option to False. Example:
from core.ftx.rest.ftx_rest_api import FtxRestApi
from core.stock.crypto_pair_manager import CryptoPairManager
ftx_rest_api: FtxRestApi = FtxRestApi()
btc_pair_manager: CryptoPairManager = CryptoPairManager("BTC-PERP", self.ftx_rest_api)
btc_pair_manager.add_time_frame(15, False)
btc_pair_manager.start_all_time_frame_acq()| Name | Purpose |
|---|---|
| twitter_elon_musk_doge_tracker | Track abnormal DOGE-PERP volume increase when Elon Musk tweet something. This strategy has an internal config/private/twitter_config.py file that has to be created before being able to run it. As for ftx_config.py, it comes with a template. Make sure to use x20 leverage on your sub account before using this strategy (account > settings > margin) |
| listing_sniper | A strategy to snipe a given pair listing by brute forcing a buy position on it until it passes. |
| multi_coin_abnormal_volume_tracker | A strategy that scan a list of pairs in order to find abnormal volume increase. For each listed coins, the strategy will compute a moving average of the last candles volume to compare it with a bigger segment of candles. |
| trend_follow | A strategy that follow a coin trend by going long when a coin price is going up, and going short when the price is going down |
| best_strat_ever | A basic strategy for testing purposes |
| demo_strategy | A basic strategy for demo purposes |
Feel free to create your own strategy in the strategy folder or reuse the existing ones. You can use FTX API examples to send trading orders. See FTX official documentation.
I'm not responsible for any money losses using this bot and guarantee that it will not do anything else that what you ask him to do in your strategies. Existing strategies can be used at your own risk, please have a deep look at the code before running it.
If you like this tool or if you're making too much money out of it, don't hesitate to give your feedback. You can share
some of your profit at my Ethereum address as well: 0xb27daa27010fc68A69b6361CCAECCE14aBEea4A8
Issues and PR are welcome
Enjoy !