How to Automate II: Backing or laying the 1st/2nd/.../nth favourite shot
This tutorial is Part two of the How to Automate series and follows on logically from the How to Automate I: Understanding Flumine tutorial we shared previously.
Make sure you look at part one, before diving into this one, as it will help you understand the general Flumine code structure and how it works. This tutorial will delve a bit deeper and give us more confidence working with Flumine. Not only will we get a cool new strategy, but we will be moving closer to learning how to automate our own model.
As always please reach out with feedback, suggestions or queries. Please submit a pull request if you catch any bugs or have other improvement suggestions!
Context
Say for example you have done some research and found there is significant published academic literature on the existence of a favourite-longshot bias. As a result, you want to automate a strategy backing all favourites or second favourites in thoroughbred racing markets. If this is the case, then this is the perfect tutorial for you!
Taking a quick look on Google Scholar we can see there is indeed plenty of published papers on this topic (almost as if I planned it). Many of these Journals are high quality too! According to the ABDC Journal Rankings:
- Scottish Journal of Political Economy has rating of A
- The Economic Journal is A*
- Applied Economics is A*
However at this point I must stress to the reader that these papers are quite old and were mainly published in the early 2000s. Schwert (2003) suggests that many of the market anomalies, discovered in financial markets as a contradiction to market efficiency, disappear once they have been published in academic literature. So, although these researchers may have found the existence of a favourite-longshot bias in betting markets in the early 2000s, they likely no longer exist as the publication of their findings leads to an increase in market efficiency.
Schwert, G.W., 2003. Anomalies and market efficiency. Handbook of the Economics of Finance, 1, pp.939-974.
Login
This is basically always the same
# Import libraries for logging in
import betfairlightweight
from flumine import Flumine, clients
# Credentials to login and logging in
trading = betfairlightweight.APIClient('username','password',app_key='appkey')
client = clients.BetfairClient(trading, interactive_login=True)
# Login
framework = Flumine(client=client)
# Code to login when using security certificates
# trading = betfairlightweight.APIClient('username','password',app_key='appkey', certs=r'C:\Users\zhoui\openssl_certs')
# client = clients.BetfairClient(trading)
# framework = Flumine(client=client)
Creating our strategy
Formulating our strategy
This is where the fun begins!
If you haven't already, read through How to Automate I: Understanding Flumine to get a grasp on what the general code structure of Flumine looks like. The basic code structure will always be the same, but we will tailor it to whatever strategy we are trying to run.
We have a few basic requirements for this strategy:
- We want to find what horse is the favourite / second favourite / third favourite and so on
- We want to place a bet on that horse
- We want to only place a single bet on that horse and on the market in general
Hold up, how do we decide which horse is the favourite?
There are two things we must consider here:
- What price should we be using to determine if a horse is the favourite?
- Back price
- Lay price
- Some sort of mid point
- Last traded price
- Some combination??
- What point in time should we reference the price to determine the favourite?
- 10 mins before the jump
- 5 mins before the jump
- 30 secs before the jump??
These decisions can make huge differences, and it's up to you to do your own analysis and decided which is best.
For this example, we will keep it simple and place a Back bet on the favourite, and use the last traded price, 60 seconds before the scheduled jump time
Implementing our strategy
Now that we have decided on using the last traded price for each horse, we actually need a way to find this in real time. Luckily, the Betfair documentation provides everything we need.
In Betfairlightweight and Flumine wrapper for the Betfair API, runner has the attribute last_price_traded, so to access it we can simply call runner.last_price_traded. Looping through all the runners in a race we can collect the last_price_traded for all runners and then convert that into a dataframe:
snapshot_last_price_traded = []
for runner in market_book.runners:
snapshot_last_price_traded.append([runner.selection_id,runner.last_price_traded])
snapshot_last_price_traded = pd.DataFrame(snapshot_last_price_traded, columns=['selection_id','last_traded_price'])
So, we end up with a DataFrame like this:
Let's sort the rows by the last_traded_price and define a new variable called the fav_selection_id which is the selection_id of the horse we want to bet on (first favourite horse).
To get the corresponding selection_id we need to select the value from the first row of the selection_id column. As Python starts indexing rows from 0, instead of 1, we will be selecting the 0th index to get the value of the first row. (If your strategy was backing/laying the second favourite you will need to be selecting the 1st index)
snapshot_last_price_traded = snapshot_last_price_traded.sort_values(by = ['last_traded_price'])
fav_selection_id = snapshot_last_price_traded['selection_id'].iloc[0]
We also need a way to validate bets so that we don't bet on multiple selections. This is because process_market_book will run every time anyone places or cancels a bet on that market. So we need to prevent multiple bet placements. This is usually easily done by setting max_trade_count = 1 however, looking at the documentation we can see that this is only on a per selection basis. As the favourite horse can change over time then we may end up betting multiple times if the favourite changes. In most situations this probably isn't an issue, as you can just limit betting to a timeframe e.g., from 60 seconds before the jump to 50 seconds before the jump. However, if the market is very illiquid there may not be any bets placed or cancelled in the time frame. So, let's come up with a way that our strategy can only bet once without using the timeframe as a work around.
According to the documentation for Flumine we can get the runner_context, which will allow us to collect info on the trades, matched and waiting to be matched. If we loop through all the runners in the market, we can turn that into a DataFrame:
runner_context = []
for runner in market_book.runners:
runner_context = self.get_runner_context(
market.market_id, runner.selection_id, runner.handicap
)
snapshot_runner_context.append([runner_context.selection_id, runner_context.executable_orders, runner_context.live_trade_count, runner_context.trade_count])
snapshot_runner_context = pd.DataFrame(snapshot_runner_context, columns=['selection_id','executable_orders','live_trade_count','trade_count'])
This will return a DataFrame like this:
Now we can simply just check if the sum last 3 columns equal zero, to validate that no bets have been placed:
Let's put this all together and pepper in some logging so we know what's happening:
# Import necessary libraries
from flumine import BaseStrategy
from flumine.order.trade import Trade
from flumine.order.order import LimitOrder, OrderStatus
from flumine.markets.market import Market
from betfairlightweight.filters import streaming_market_filter
from betfairlightweight.resources import MarketBook
import pandas as pd
import numpy as np
# Logging
import logging
logging.basicConfig(filename = 'how_to_automate_2.log', level=logging.INFO, format='%(asctime)s:%(levelname)s:%(lineno)d:%(message)s')
class BackFavStrategy(BaseStrategy):
# Defines what happens when we start our strategy i.e. this method will run once when we first start running our strategy
def start(self) -> None:
print("starting strategy 'BackFavStrategy'")
def check_market_book(self, market: Market, market_book: MarketBook) -> bool:
# process_market_book only executed if this returns True
if market_book.status != "CLOSED":
return True
def process_market_book(self, market: Market, market_book: MarketBook) -> None:
# Collect data on last price traded and the number of bets we have placed
snapshot_last_price_traded = []
snapshot_runner_context = []
for runner in market_book.runners:
snapshot_last_price_traded.append([runner.selection_id,runner.last_price_traded])
# Get runner context for each runner
runner_context = self.get_runner_context(
market.market_id, runner.selection_id, runner.handicap
)
snapshot_runner_context.append([runner_context.selection_id, runner_context.executable_orders, runner_context.live_trade_count, runner_context.trade_count])
# Convert last price traded data to dataframe
snapshot_last_price_traded = pd.DataFrame(snapshot_last_price_traded, columns=['selection_id','last_traded_price'])
# Find the selection_id of the favourite
snapshot_last_price_traded = snapshot_last_price_traded.sort_values(by = ['last_traded_price'])
fav_selection_id = snapshot_last_price_traded['selection_id'].iloc[0]
logging.info(snapshot_last_price_traded) # logging
# Convert data on number of bets we have placed to a dataframe
snapshot_runner_context = pd.DataFrame(snapshot_runner_context, columns=['selection_id','executable_orders','live_trade_count','trade_count'])
logging.info(snapshot_runner_context) # logging
for runner in market_book.runners:
if runner.status == "ACTIVE" and market.seconds_to_start < 60 and market_book.inplay == False and runner.selection_id == fav_selection_id and snapshot_runner_context.iloc[:,1:].sum().sum() == 0:
trade = Trade(
market_id=market_book.market_id,
selection_id=runner.selection_id,
handicap=runner.handicap,
strategy=self,
)
order = trade.create_order(
side="BACK", order_type=LimitOrder(price=runner.last_price_traded, size=5)
)
market.place_order(order)
Running our strategy
Now that we have our strategy ready, we can point it to a sport, and let it run. This time we will add some trading controls because we are likely to get matched. Let's specify that we are only comfortable with 1 bet at any time with a maximum exposure of $20
strategy = BackFavStrategy(
market_filter=streaming_market_filter(
event_type_ids=["4339"], # Greyhounds
country_codes=["AU"], # Australian Markets
market_types=["WIN"], # Win Markets
),
max_trade_count=1, # max total number of trades per runner
max_live_trade_count=1, # max live (with executable orders) trades per runner
max_selection_exposure=20, # max exposure of 20 per horse
max_order_exposure= 20 # Max bet sizes of $20
)
Before we start running our strategy lets learn how the Flumine framework actually works. I basically glossed over this in Part I as its not entirely necessary to get something up and running, but in a more realistic sense its very useful and we now have a good strategy to test it on.
When we log into using Flumine we define a framework which is a Flumine object:
trading = betfairlightweight.APIClient('username','password',app_key='appkey')
client = clients.BetfairClient(trading, interactive_login=True)
framework = Flumine(client=client)
When we run our strategies, we are actually running framework. You can think of framework as a video game character with nothing equiped, when we add different strategies its like equipping a weapon like a sword or bow. We can also add other things to help us out such as LiveLoggingControl (which we will learn to create a bit futher down), and an autoterminate function so we don't need to manually turn it off each day, this is like adding buffs or armour that can help our character to progress.
We need to define the thing that we are adding onto our framework and then Flumine will have specific functions that allow us to equip our strategies and helper functions. For example to add a strategy we just need to do add_strategy(), the list of Flumine functions that you can use to add things to framework are available here. Once we have our character ready with strategies and help code equiped we can do framework.run() and that will run your framework with all the strategies and all the supporting supporting code attached.
Automatic Terminate
This code is a direct copy and paste from the examples and works like a charm out of the box. It runs every 60 seconds and checks if all markets starting today have been closed for at least 20 minutes. If it has then it will stop our automation.
import logging
import datetime
from flumine.worker import BackgroundWorker
from flumine.events.events import TerminationEvent
logger = logging.getLogger(__name__)
"""
Worker can be used as followed:
framework.add_worker(
BackgroundWorker(
framework,
terminate,
func_kwargs={"today_only": True, "seconds_closed": 1200},
interval=60,
start_delay=60,
)
)
This will run every 60s and will terminate
the framework if all markets starting 'today'
have been closed for at least 1200s
"""
# Function that stops automation running at the end of the day
def terminate(
context: dict, flumine, today_only: bool = True, seconds_closed: int = 600
) -> None:
"""terminate framework if no markets
live today.
"""
# Creates a list of all markets
markets = list(flumine.markets.markets.values())
# from the above list, create a list of markets that are expected to start today
markets_today = [
m
for m in markets
if m.market_start_datetime.date() == datetime.datetime.utcnow().date()
and (
m.elapsed_seconds_closed is None
or (m.elapsed_seconds_closed and m.elapsed_seconds_closed < seconds_closed)
)
]
# counts the markets that are expected to start today
if today_only:
market_count = len(markets_today)
else:
market_count = len(markets)
# if the number of markets that are expected to start today then stop flumine
if market_count == 0:
logger.info("No more markets available, terminating framework")
flumine.handler_queue.put(TerminationEvent(flumine))
Now that we have created our terminate function we have to add it to our framework
Let's also create something that records our bets in a nice csv/excel file so we can review how we went later on. Although this is also called logging it will create a clean csv file called "orders_hta_2.csv" that looks like this:
This becomes super useful when we have more than one strategy so we can track how each strategy is performing (it also reads nicely into a pandas DataFrame!). This will also allow us to do some analysis such as check how often our bets get matched. This code is copied from the examples with some very slight changes.
import os
import csv
import logging
from flumine.controls.loggingcontrols import LoggingControl
from flumine.order.ordertype import OrderTypes
logger = logging.getLogger(__name__)
FIELDNAMES = [
"bet_id",
"strategy_name",
"market_id",
"selection_id",
"trade_id",
"date_time_placed",
"price",
"price_matched",
"size",
"size_matched",
"profit",
"side",
"elapsed_seconds_executable",
"order_status",
"market_note",
"trade_notes",
"order_notes",
]
class LiveLoggingControl(LoggingControl):
NAME = "BACKTEST_LOGGING_CONTROL"
def __init__(self, *args, **kwargs):
super(LiveLoggingControl, self).__init__(*args, **kwargs)
self._setup()
# checks if the file "orders_hta_2.csv" already exists, if it doens't then create it
def _setup(self):
if os.path.exists("orders_hta_2.csv"):
logging.info("Results file exists")
else:
with open("orders_hta_2.csv", "w") as m:
# Create orders_hta_2.csv with the first row as the FIELDNAMES we specified above
csv_writer = csv.DictWriter(m, delimiter=",", fieldnames=FIELDNAMES)
csv_writer.writeheader()
def _process_cleared_orders_meta(self, event):
orders = event.event # gives us a list of our orders for a market that has already settled
with open("orders_hta_2.csv", "a") as m: # open orders_hta_2.csv and append a new row of data (orders)
for order in orders:
if order.order_type.ORDER_TYPE == OrderTypes.LIMIT:
size = order.order_type.size
else:
size = order.order_type.liability
if order.order_type.ORDER_TYPE == OrderTypes.MARKET_ON_CLOSE:
price = None
else:
price = order.order_type.price
try: # Create a dictionary of data we want to append to our csv file
order_data = {
"bet_id": order.bet_id,
"strategy_name": order.trade.strategy,
"market_id": order.market_id,
"selection_id": order.selection_id,
"trade_id": order.trade.id,
"date_time_placed": order.responses.date_time_placed,
"price": price,
"price_matched": order.average_price_matched,
"size": size,
"size_matched": order.size_matched,
"profit": 0 if not order.cleared_order else order.cleared_order.profit,
"side": order.side,
"elapsed_seconds_executable": order.elapsed_seconds_executable,
"order_status": order.status.value,
"market_note": order.trade.market_notes,
"trade_notes": order.trade.notes_str,
"order_notes": order.notes_str,
}
csv_writer = csv.DictWriter(m, delimiter=",", fieldnames=FIELDNAMES) # maps our dictionary to output rows
csv_writer.writerow(order_data) # append data to csv files
except Exception as e:
logger.error(
"_process_cleared_orders_meta: %s" % e,
extra={"order": order, "error": e},
)
logger.info("Orders updated", extra={"order_count": len(orders)})
def _process_cleared_markets(self, event):
cleared_markets = event.event
for cleared_market in cleared_markets.orders:
logger.info(
"Cleared market",
extra={
"market_id": cleared_market.market_id,
"bet_count": cleared_market.bet_count,
"profit": cleared_market.profit,
"commission": cleared_market.commission,
},
)
Now let's add the bet logging to our framework and run everything at once:
There are some other things you can add such as workers that run automatically in the background at set times (e.g. every 10 seconds) and are independent of market updates. There is some documentation available but to be honest I've never had to use them. If you are doing something that is a bit more intense or needs to be run at set time intervals then they are probably useful so take a look.
Conclusion and next steps
We have so far done zero backtesting on this strategy, and blindly following strategies from published papers that are over 20 years old is a sure fire way to lose money. But hopefully this gives you an idea of the things you can accomplish with Flumine. If you are keen on backtesting your own betting angles, I would suggest taking a look at our tutorial which goes into depth on how to backtest Automated Betting Angles in Python using historical data.
Now that we have a better understanding of Flumine we are getting very close to automating our own model. We still have three parts remaining in this series which will take you step by step through:
Complete code
Run the code from your ide by using py <filename>.py, making sure you amend the path to point to your input data.
# Import libraries for logging in
import betfairlightweight
from flumine import Flumine, clients
from flumine import BaseStrategy
from flumine.order.trade import Trade
from flumine.order.order import LimitOrder, OrderStatus
from flumine.markets.market import Market
from betfairlightweight.filters import streaming_market_filter
from betfairlightweight.resources import MarketBook
import pandas as pd
import numpy as np
import logging
import datetime
from flumine.worker import BackgroundWorker
from flumine.events.events import TerminationEvent
import os
import csv
import logging
from flumine.controls.loggingcontrols import LoggingControl
from flumine.order.ordertype import OrderTypes
logging.basicConfig(filename = 'how_to_automate_2.log', level=logging.INFO, format='%(asctime)s:%(levelname)s:%(lineno)d:%(message)s')
# Import libraries for logging in
import betfairlightweight
from flumine import Flumine, clients
# Credentials to login and logging in
trading = betfairlightweight.APIClient('username','password',app_key='appkey')
client = clients.BetfairClient(trading, interactive_login=True)
# Login
framework = Flumine(client=client)
# Code to login when using security certificates
# trading = betfairlightweight.APIClient('username','password',app_key='appkey', certs=r'C:\Users\zhoui\openssl_certs')
# client = clients.BetfairClient(trading)
# framework = Flumine(client=client)
class BackFavStrategy(BaseStrategy):
# Defines what happens when we start our strategy i.e. this method will run once when we first start running our strategy
def start(self) -> None:
print("starting strategy 'BackFavStrategy'")
def check_market_book(self, market: Market, market_book: MarketBook) -> bool:
# process_market_book only executed if this returns True
if market_book.status != "CLOSED":
return True
def process_market_book(self, market: Market, market_book: MarketBook) -> None:
# Collect data on last price traded and the number of bets we have placed
snapshot_last_price_traded = []
snapshot_runner_context = []
for runner in market_book.runners:
snapshot_last_price_traded.append([runner.selection_id,runner.last_price_traded])
# Get runner context for each runner
runner_context = self.get_runner_context(
market.market_id, runner.selection_id, runner.handicap
)
snapshot_runner_context.append([runner_context.selection_id, runner_context.executable_orders, runner_context.live_trade_count, runner_context.trade_count])
# Convert last price traded data to dataframe
snapshot_last_price_traded = pd.DataFrame(snapshot_last_price_traded, columns=['selection_id','last_traded_price'])
# Find the selection_id of the favourite
snapshot_last_price_traded = snapshot_last_price_traded.sort_values(by = ['last_traded_price'])
fav_selection_id = snapshot_last_price_traded['selection_id'].iloc[0]
logging.info(snapshot_last_price_traded) # logging
# Convert data on number of bets we have placed to a dataframe
snapshot_runner_context = pd.DataFrame(snapshot_runner_context, columns=['selection_id','executable_orders','live_trade_count','trade_count'])
logging.info(snapshot_runner_context) # logging
for runner in market_book.runners:
if runner.status == "ACTIVE" and market.seconds_to_start < 60 and market_book.inplay == False and runner.selection_id == fav_selection_id and snapshot_runner_context.iloc[:,1:].sum().sum() == 0:
trade = Trade(
market_id=market_book.market_id,
selection_id=runner.selection_id,
handicap=runner.handicap,
strategy=self,
)
order = trade.create_order(
side="BACK", order_type=LimitOrder(price=runner.last_price_traded, size=5)
)
market.place_order(order)
logger = logging.getLogger(__name__)
"""
Worker can be used as followed:
framework.add_worker(
BackgroundWorker(
framework,
terminate,
func_kwargs={"today_only": True, "seconds_closed": 1200},
interval=60,
start_delay=60,
)
)
This will run every 60s and will terminate
the framework if all markets starting 'today'
have been closed for at least 1200s
"""
# Function that stops automation running at the end of the day
def terminate(
context: dict, flumine, today_only: bool = True, seconds_closed: int = 600
) -> None:
"""terminate framework if no markets
live today.
"""
markets = list(flumine.markets.markets.values())
markets_today = [
m
for m in markets
if m.market_start_datetime.date() == datetime.datetime.utcnow().date()
and (
m.elapsed_seconds_closed is None
or (m.elapsed_seconds_closed and m.elapsed_seconds_closed < seconds_closed)
)
]
if today_only:
market_count = len(markets_today)
else:
market_count = len(markets)
if market_count == 0:
logger.info("No more markets available, terminating framework")
flumine.handler_queue.put(TerminationEvent(flumine))
FIELDNAMES = [
"bet_id",
"strategy_name",
"market_id",
"selection_id",
"trade_id",
"date_time_placed",
"price",
"price_matched",
"size",
"size_matched",
"profit",
"side",
"elapsed_seconds_executable",
"order_status",
"market_note",
"trade_notes",
"order_notes",
]
class LiveLoggingControl(LoggingControl):
NAME = "BACKTEST_LOGGING_CONTROL"
def __init__(self, *args, **kwargs):
super(LiveLoggingControl, self).__init__(*args, **kwargs)
self._setup()
# Changed file path and checks if the file orders_hta_2.csv already exists, if it doens't then create it
def _setup(self):
if os.path.exists("orders_hta_2.csv"):
logging.info("Results file exists")
else:
with open("orders_hta_2.csv", "w") as m:
csv_writer = csv.DictWriter(m, delimiter=",", fieldnames=FIELDNAMES)
csv_writer.writeheader()
def _process_cleared_orders_meta(self, event):
orders = event.event
with open("orders_hta_2.csv", "a") as m:
for order in orders:
if order.order_type.ORDER_TYPE == OrderTypes.LIMIT:
size = order.order_type.size
else:
size = order.order_type.liability
if order.order_type.ORDER_TYPE == OrderTypes.MARKET_ON_CLOSE:
price = None
else:
price = order.order_type.price
try:
order_data = {
"bet_id": order.bet_id,
"strategy_name": order.trade.strategy,
"market_id": order.market_id,
"selection_id": order.selection_id,
"trade_id": order.trade.id,
"date_time_placed": order.responses.date_time_placed,
"price": price,
"price_matched": order.average_price_matched,
"size": size,
"size_matched": order.size_matched,
"profit": 0 if not order.cleared_order else order.cleared_order.profit,
"side": order.side,
"elapsed_seconds_executable": order.elapsed_seconds_executable,
"order_status": order.status.value,
"market_note": order.trade.market_notes,
"trade_notes": order.trade.notes_str,
"order_notes": order.notes_str,
}
csv_writer = csv.DictWriter(m, delimiter=",", fieldnames=FIELDNAMES)
csv_writer.writerow(order_data)
except Exception as e:
logger.error(
"_process_cleared_orders_meta: %s" % e,
extra={"order": order, "error": e},
)
logger.info("Orders updated", extra={"order_count": len(orders)})
def _process_cleared_markets(self, event):
cleared_markets = event.event
for cleared_market in cleared_markets.orders:
logger.info(
"Cleared market",
extra={
"market_id": cleared_market.market_id,
"bet_count": cleared_market.bet_count,
"profit": cleared_market.profit,
"commission": cleared_market.commission,
},
)
strategy = BackFavStrategy(
market_filter=streaming_market_filter(
event_type_ids=["4339"], # Greyhounds
country_codes=["AU"], # Australian Markets
market_types=["WIN"], # Win Markets
),
max_trade_count=1, # max total number of trades per runner
max_live_trade_count=1, # max live (with executable orders) trades per runner
max_selection_exposure=20, # max exposure of 20 per horse
max_order_exposure= 20 # Max bet sizes of $20
)
framework.add_strategy(strategy)
# Add the auto terminate to our framework
framework.add_worker(
BackgroundWorker(
framework,
terminate,
func_kwargs={"today_only": True, "seconds_closed": 1200},
interval=60,
start_delay=60,
)
)
framework.add_logging_control(
LiveLoggingControl()
)
framework.run() # run our framework
Disclaimer
Note that whilst models and automated strategies are fun and rewarding to create, we can't promise that your model or betting strategy will be profitable, and we make no representations in relation to the code shared or information on this page. If you're using this code or implementing your own strategies, you do so entirely at your own risk and you are responsible for any winnings/losses incurred. Under no circumstances will Betfair be liable for any loss or damage you suffer.