Interactive Brokers is a discount brokerage that provides a good API for programatically accessing their platform. The purpose of this post is to create an application that will capture tick level data and save that data into a database for future use.
I started to use the IBrokers package in R to do this post. However, as R is NOT easily threaded and the IB API is heavily threaded, well... oil and water.
Instead I went with the C# port of the API from
DinosaurTech. It's good and it's free.
Luckily you do not need an account with Interactive Brokers for this project. They have a
demo environment available on their webpage. The data is FAKE, but it's good enough to test connectivity. Simply run the demo, then go to
Configure->API->Settings and insure that the "Enable ActiveX and Socket Clients" is checked and the port is set to 7496.
To follow along in this post, you will need
- MySQL, MySQL Workbench, and MySQL Connector for .NET (all available here).
- VS2010 with C#. You can download the free Express version here.
- The libraries from DinosaurTech available at the link above.
- A basic understanding of C# (available here [Pro C# 2010 and the .NET 4 Platform]
, here [Beginning Visual C# 2010 (Wrox Programmer to Programmer)]
, or here[Google] ).
To start, install all of the above items. In MySQL workbench, run the following SQL to create the table to store ticks.
SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0;
SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0;
SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='TRADITIONAL';
DROP SCHEMA IF EXISTS `tick_db` ;
CREATE SCHEMA IF NOT EXISTS `tick_db` DEFAULT CHARACTER SET latin1
COLLATE latin1_swedish_ci ;
USE `tick_db` ;
-- -----------------------------------------------------
-- Table `ticks`
-- -----------------------------------------------------
DROP TABLE IF EXISTS `ticks` ;
CREATE TABLE IF NOT EXISTS
`ticks` (
`idticks` INT NOT NULL ,
`symbol` VARCHAR(8) NOT NULL ,
`date` DATE NOT NULL ,
`time` TIME NOT NULL ,
`value` FLOAT NOT NULL ,
`type` VARCHAR(12) NOT NULL ,
PRIMARY KEY (`idticks`, `date`)
)
ENGINE = InnoDB PARTITION BY KEY(date) PARTITIONS 1;
CREATE INDEX `Symbol` ON `ticks` (`symbol` ASC) ;
SET SQL_MODE=@OLD_SQL_MODE;
SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS;
SET UNIQUE_CHECKS=@OLD_UNIQUE_CHECKS;
You will note that we are partitioning the table by date. The goal of this is so that as the data grows, day over day, that we can still query quickly.
If you do not not have a user in the database, create one and give it permissions on this table (my user is dom and the password, a super secure dom123).
Before we go on, let's discuss how the IB API works. Your programming environment connects through their trading platform called TWS (Trader Work Station) as it is running on your computer. It uses the TWS connections to the IB servers to make your requests. When you request market tick data for a stock, the data arrives whenever it wants. There is no synchronous get() function. You ask for data and the API fires events when that data comes back. In this, you can have multiple requests running at the same time. You just have to keep straight what equity data you are processing at any given time.
This threaded nature means that we can process a large number of requests. It also means that we need to free up the threads that are receiving data as quickly as possible. I.E. they shouldn't be writing to the database. We will create a queue of ticks in a separate thread that will post those to the database as quickly as it can.
Ticks are reported as updates to BID, BID_SIZE, ASK, ASK_SIZE, LAST, LAST_SIZE, VOLUME, HIGH, LOW, and CLOSE.
Here is the plan of action:
- Create a new console application in C#.
- Create a Tick class for holding the tick and give it a method that returns the SQL INSERT statement to put that tick into our database.
- Create a TickQueue class that will process ticks that are inserted into its Queue object as fast as possible.
- Create a TickMain class that will request all the market data and insert the tick events into the TickQueue.
- Call the TickMain class from a separate thread in the console application and watch the magic happen.
First, our Tick class
public class Tick
{
public String type
= "";
public decimal value
= 0;
public String date
= System.DateTime.Now.ToString("yyyy-MM-dd");
public String time
= System.DateTime.Now.ToString("H:mm:ss.ss");
public String
symbol = "";
public Int32 index
= 0;
public Tick(String
type, decimal value, String
symbol, Int32 index)
{
this.type = type;
this.value = value;
this.symbol = symbol;
this.index = index;
}
public Tick() { }
public String
toInsert()
{
String output = "insert
into ticks (idticks,symbol,date,time,value,type) values (" +
index +
",'" + symbol +
"', DATE('" + date +
"'),TIME('" + time + "')," +
value +
",'" + type + "')";
return output;
}
}
This class is basic. We simply hold the information about the tick, and create a method that will create our SQL INSERT statement.
Next is the TickQueue. This is more involved...
public class TickQueue
{
// Internal queue of db inserts
private Queue<Tick> queue = new
Queue<Tick>();
// Locking Object.
Necessary as multiple threads will be
// accessing this object simultanously
private Object
qLockObj = new object();
//Connection object to MySQL
public MySqlConnection
conn = null;
//Flag to stop processing
public bool stop = false;
//Method to enqueue a write
public void Add(Tick item)
{
//Lock for single access only
lock(qLockObj)
{
queue.Enqueue(item);
}
}
//Method to write items as they are enqueued
public void Run()
{
int n = 0;
//Loop while the stop flag is false
while (!stop)
{
//Lock and get a count of the object in the queue
lock (qLockObj)
{
n = queue.Count;
}
//If there are objects in the queue, then process them
if (n > 0)
{
process();
}
//Sleep for .1 seconds before looping again
System.Threading.Thread.Sleep(100);
}
//When the shutdown flag is received, write any
//values still in the queue and then stop
Console.WriteLine("Shutting
Down TickQueue; " + queue.Count + "
items left");
process();
}
//Method to process items in the queue
private void
process()
{
List<Tick>
inserts = new List<Tick>();
int i = 0;
//Loop through the items in the queue and put them in a
list
lock (qLockObj)
{
for (i = 0; i < queue.Count; i++)
inserts.Add(queue.Dequeue());
}
Console.WriteLine("Processing
" + i + " items into
database");
//call insert for each item.
foreach (Tick t in inserts)
insert(t);
}
//Method to insert a tick
private void insert(Tick t)
{
using (MySqlCommand
cmd = conn.CreateCommand())
{
cmd.CommandText = t.toInsert();
try
{
cmd.ExecuteNonQuery();
}
catch (Exception
exp)
{
Console.WriteLine("OOPS " + exp.Message);
}
}
}
}
You will notice that I am not using threads in this object. The TickMain object will have the worker thread that call the Run() method in TickQueue. Because multiple threads will be accessing the object, I've surrounded all access points to the actual queue with a lock() statement. That will insure that only 1 thread at a time gets access.
I've tried to comment enough to give an idea of what is happening in there. If not, let me know and I will expand. The same goes for TickMain below.
Now the TickMain class:
public class TickMain
{
//Private and public accessor for the IBClient object
private IBClient
_client = null;
public IBClient
client
{
get { return _client;
}
set
{
_client = value;
//If the client is connected, then set the queue.stop =
false
if (_client.Connected)
queue.stop = false;
else
queue.stop = true;
}
}
public List<String> stockList = new
List<string>();
public MySqlConnection
conn = null;
public bool doGet = true;
private TickQueue
queue = new TickQueue();
private BackgroundWorker
bg = new BackgroundWorker();
private Dictionary<int, String>
tickId = new Dictionary<int, string>();
private int tickIndex
= 0;
private object
lockObj = new object();
//Constructors
public TickMain()
{
initialize();
}
public TickMain(IBClient
client, List<String>
stockList, MySqlConnection conn)
{
this.client = client;
this.stockList = stockList;
this.conn = conn;
initialize();
}
//Initialization method
private void
initialize()
{
//Setup the background worker to run the queue
bg.DoWork += new DoWorkEventHandler(bg_DoWork);
//Connect to MySQL if we haven't already
if (conn.State != System.Data.ConnectionState.Open)
conn.Open();
//Don't process the queue if not connected to IB
if (!client.Connected)
queue.stop = true;
//Set the MySQL connection for hte queue
queue.conn = conn;
//Get the next value of the queue index
using (MySqlCommand
cmd = conn.CreateCommand())
{
cmd.CommandText = "select
coalesce(max(idticks),0) from ticks";
MySqlDataReader Reader;
Reader = cmd.ExecuteReader();
Reader.Read();
tickIndex = Reader.GetInt32(0) + 1;
Reader.Close();
}
}
//Method for getting market prices
public void Run()
{
if (client.Connected)
{
//Set up the event handlers for the ticks
client.TickPrice += new EventHandler<TickPriceEventArgs>(client_TickPrice);
client.TickSize += new
EventHandler<TickSizeEventArgs>(client_TickSize);
//Initialize a counter for stock symbols
int i = 1;
//Start the queue
bg.RunWorkerAsync();
//Request market
data for each stock in the stockList
foreach (String
str in stockList)
{
tickId.Add(i, str);
client.RequestMarketData(i, new Equity(str), null,
false, false);
i++;
}
//Hang out until told otherwise
while (doGet)
{
System.Threading.Thread.Sleep(100);
}
//Remove event handlers
Console.WriteLine("Shutting
Down TickMain");
client.TickPrice -= new EventHandler<TickPriceEventArgs>(client_TickPrice);
client.TickSize -= new EventHandler<TickSizeEventArgs>(client_TickSize);
queue.stop = true;
}
}
//Event handler for TickSize events
void client_TickSize(object
sender, TickSizeEventArgs e)
{
//Get the symbol from the dictionary
String symbol = tickId[e.TickerId];
int i = 0;
//As this is asynchronous, lock and get the current tick
index
lock (lockObj)
{
i
= tickIndex;
tickIndex++;
}
//Create a tick object and enqueue it
Tick tick = new Tick(EnumDescConverter.GetEnumDescription(e.TickType),
e.Size, symbol, i);
queue.Add(tick);
}
//Event Handler for TickPrice events
void client_TickPrice(object
sender, TickPriceEventArgs e)
{
//Get the symbol from the dictionary
String
symbol = tickId[e.TickerId];
int i = 0;
//As this is asynchronous, lock and get the current tick
index
lock (lockObj)
{
i
= tickIndex;
tickIndex++;
}
//Create a tick object and enqueue it
Tick tick = new Tick(EnumDescConverter.GetEnumDescription(e.TickType),
e.Price, symbol, i);
queue.Add(tick);
}
//BackgroundWorker delegate to run the queue.
private void
bg_DoWork(object sender, DoWorkEventArgs e)
{
queue.Run();
}
}
Stock requests are given a unique ID. The tick events have this ID, not the symbol. So we keep track of the ID and symbol pairs in a Dictionary. You will note that we have a BackgroundWorker in here that calls and run the database write queue. The tick event handlers process each tick, assign it a unique index ID (another source of possible thread contention, hence the lock() around the index creation).
Finally the program Class in the console application
class Program
{
static MySqlConnection
conn =
new MySqlConnection("server=LOCALHOST;DATABASE=tick_db;USER=dom;PASSWORD=dom123");
static TickMain
main = null;
static void Main(string[] args)
{
//Open the connections.
conn.Open();
IBClient client = new
IBClient();
client.Connect("localhost",
7496, 2);
//List of stock ticks to get
List<String>
stockList = new List<string>();
stockList.Add("GOOG");
stockList.Add("SPY");
stockList.Add("SH");
stockList.Add("DIA");
//Initialize the TickMain object
main
= new TickMain(client,
stockList, conn);
main.doGet = true;
//Setup a worker to call main.Run() asycronously.
BackgroundWorker bg = new
BackgroundWorker();
bg.DoWork += new DoWorkEventHandler(bg_DoWork);
bg.RunWorkerAsync();
//Chill until the user hits enter then stop the TickMain
object
Console.ReadLine();
main.doGet = false;
//disconnect
client.Disconnect();
Console.WriteLine("Hit
Enter to Continue");
Console.ReadLine();
}
//Delegate for main.Run()
static void bg_DoWork(object
sender, DoWorkEventArgs e)
{
main.Run();
}
}
The application will run, scrolling updates on how many records are being written to the database until you hit the enter key. After that, the system shuts down and you are prompted to hit enter once more to exit the application.
The entire VS2010 project and SQL for the table can be found
here.
My plan is to set this up to run all next week with live data, pull it into R next weekend, and see what we can see.
