Calculating OHLC bars

Function bar is generally used to group data based on specified interval.

date = 09:32m 09:33m 09:45m 09:49m 09:56m 09:56m;
bar(date, 5);

The result is:

[09:30m,09:30m,09:45m,09:45m,09:55m,09:55m]

Example 1:

n = 1000000
date = take(2019.11.07 2019.11.08, n)
time = (09:30:00.000 + rand(int(6.5*60*60*1000), n)).sort!()
timestamp = concatDateTime(date, time)
price = 100+cumsum(rand(0.02, n)-0.01)
volume = rand(1000, n)
symbol = rand(`AAPL`FB`AMZN`MSFT, n)
trade = table(symbol, date, time, timestamp, price, volume).sortBy!(`symbol`timestamp)
undef(`date`time`timestamp`price`volume`symbol);

Calculate 5-minute OHLC bars:

barMinutes = 5m
OHLC = select first(price) as open, max(price) as high, min(price) as low, last(price) as close, sum(volume) as volume from trade group by symbol, date, bar(time, barMinutes) as barStart

Parameter interval of function bar supports the DURATION type. "5m" here means 5 minutes.

Use function dailyAlignedBar to specify the starting time of OHLC windows. This function can accommodate multiple trading sessions per day as well as overnight sessions.

Note that for function dailyAlignedBar, the data type of the temporal column can be SECOND, TIME, NANOTIME, DATETIME, TIMESTAMP and NANOTIMESTAMP. The parameter 'timeOffset' that specifies the starting time of each trading session must have the corresponding data type: SECOND, TIME or NANOTIME, respectively.

Example 2 (one trading session per day): Calculate 7-minute OHLC bars with the same table 'trade' in Example 1.

barMinutes = 7m
OHLC = select first(price) as open, max(price) as high, min(price) as low, last(price) as close, sum(volume) as volume from trade group by symbol, dailyAlignedBar(timestamp, 09:30:00.000, barMinutes) as barStart

Example 3 (two trading sessions per day): China's stock market has two trading sessions per day, from 9:30 to 11:30 in the morning and from 13:00 to 15:00 in the afternoon.

Use the following script to generate simulated data:

n = 1000000
date = take(2019.11.07 2019.11.08, n)
time = (09:30:00.000 + rand(2*60*60*1000, n/2)).sort!() join (13:00:00.000 + rand(2*60*60*1000, n/2)).sort!()
timestamp = concatDateTime(date, time)
price = 100+cumsum(rand(0.02, n)-0.01)
volume = rand(1000, n)
symbol = rand(`600519`000001`600000`601766, n)
trade = table(symbol, timestamp, price, volume).sortBy!(`symbol`timestamp)
undef(`date`time`timestamp`price`volume`symbol)

Calculate the 7-minute OHLC bars:

barMinutes = 7m
sessionsStart=09:30:00.000 13:00:00.000
sessionsEnd=11:30:00.000 15:00:00.000
OHLC = select first(price) as open, max(price) as high, min(price) as low, last(price) as close, sum(volume) as volume from trade group by symbol, dailyAlignedBar(timestamp, sessionsStart, barMinutes, sessionsEnd) as barStart

Example 4 (two trading sessions per day with an overnight session): Some futures have multiple trading sessions per day including an overnight session. In this example, the first trading session is from 8:45 AM to 13:45 PM, and the other session is an overnight session from 15:00 PM to 05:00 AM the next day.

Use the following script to generate simulated data:

daySession =  08:45:00.000 : 13:45:00.000
nightSession = 15:00:00.000 : 05:00:00.000
n = 1000000
timestamp = rand(concatDateTime(2019.11.06, daySession[0]) .. concatDateTime(2019.11.08, nightSession[1]), n).sort!()
price = 100+cumsum(rand(0.02, n)-0.01)
volume = rand(1000, n)
symbol = rand(`A120001`A120002`A120003`A120004, n)
trade = select * from table(symbol, timestamp, price, volume) where timestamp.time() between daySession or timestamp.time()>=nightSession[0] or timestamp.time()<nightSession[1] order by symbol, timestamp
undef(`timestamp`price`volume`symbol);

Calculate the 7-minute OHLC bars:

barMinutes = 7m
sessionsStart = [daySession[0], nightSession[0]]
OHLC = select first(price) as open, max(price) as high, min(price) as low, last(price) as close, sum(volume) as volume from trade group by symbol, dailyAlignedBar(timestamp, sessionsStart, barMinutes) as barStart

In the examples above, the OHLC windows do not overlap. To calculate overlapping OHLC windows, you can use function wj. With the wj function, each row in the left table corresponds to a window in the right table, and calculations can be conducted on this window.

Example 5. 2 trading sessions per day with overlapping OHLC windows

Simulate Chinese stock market data and calculate a 30-minute OHLC bars every 5 minutes.

n = 1000000
sampleDate = 2019.11.07
symbols = `600519`000001`600000`601766
trade = table(take(sampleDate, n) as date, 
	(09:30:00.000 + rand(7200000, n/2)).sort!() join (13:00:00.000 + rand(7200000, n/2)).sort!() as time, 
	rand(symbols, n) as symbol, 
	100+cumsum(rand(0.02, n)-0.01) as price, 
	rand(1000, n) as volume)

First generate OHLC windows, then use function cj(cross join) to generate a combination of stock symbols and OHLC windows.

barWindows = table(symbols as symbol).cj(table((09:30:00.000 + 0..23 * 300000).join(13:00:00.000 + 0..23 * 300000) as time))

Then use function wj to calculate OHLC bars with overlapping windows:

OHLC = wj(barWindows, trade, 0:(30*60*1000), 
		<[first(price) as open, max(price) as high, min(price) as low, last(price) as close, sum(volume) as volume]>, `symbol`time)

The windows in all of the examples above were determined with time. You can use other variables such as trading volume as the basis to determine windows.

Example 6: OHLC bars are calculated every time trading volume increases by 1,000,000.

n = 1000000
sampleDate = 2019.11.07
symbols = `600519`000001`600000`601766
trade = table(take(sampleDate, n) as date, 
	(09:30:00.000 + rand(7200000, n/2)).sort!() join (13:00:00.000 + rand(7200000, n/2)).sort!() as time, 
	rand(symbols, n) as symbol, 
	100+cumsum(rand(0.02, n)-0.01) as price, 
	rand(1000, n) as volume)
	
volThreshold = 1000000
t = select first(time) as barStart, first(price) as open, max(price) as high, min(price) as low, last(price) as close, last(cumvol) as cumvol 
from (select symbol, time, price, cumsum(volume) as cumvol from trade context by symbol)
group by symbol, bar(cumvol, volThreshold) as volBar;

If you need to extract large-scale historical data from the database, calculate OHLC bars, and then save them into the database, we can use the built-in MapReduce function mr for parallel loading and calculation. This method can significantly increase speed.

This example uses US stock market trading data. The raw data is stored in table 'trades' in database "dfs://TAQ" with a composite partition: a value partition based on trading dates and a range partition based on stock symbols.

(1) Load the metadata of the table on disk into memory:

login(`admin, `123456)
db = database("dfs://TAQ")
trades = db.loadTable("trades")

(2) Create a template table 'model', then create an empty table 'OHLC' in database "dfs://TAQ" based on the schema of the template table to store the results:

model=select top 1 Symbol, Date, Time.second() as bar, PRICE as open, PRICE as high, PRICE as low, PRICE as close, SIZE as volume from trades where Date=2007.08.01, Symbol=`EBAY
if(existsTable("dfs://TAQ", "OHLC"))
	db.dropTable("OHLC")
db.createPartitionedTable(model, `OHLC, `Date`Symbol)

(3) Use function mr to calculate OHLC bars and write the results to table 'OHLC':

def calcOHLC(inputTable){
	tmp=select first(PRICE) as open, max(PRICE) as high, min(PRICE) as low, last(PRICE) as close, sum(SIZE) as volume from inputTable where Time.second() between 09:30:00 : 15:59:59 group by Symbol, Date, 09:30:00+bar(Time.second()-09:30:00, 5*60) as bar
	loadTable("dfs://TAQ", `OHLC).append!(tmp)
	return tmp.size()
}
ds = sqlDS(<select Symbol, Date, Time, PRICE, SIZE from trades where Date between 2007.08.01 : 2019.08.01>)
mr(ds, calcOHLC, +)

• 'ds' is a series of data sources generated by function sqlDS. Each data source represents data from a partition.
• Function calcOHLC is the map function in MapReduce. It calculates OHLC bars from each data source, writes the result to the database and returns the number of rows written to the database.
• "+" is the reduce function in MapReduce. It adds up the results of all map functions (the number of rows written to the database) to return the total number of rows written to the database.

• Create a stream table in DolphinDB

share streamTable(100:0, `Symbol`Datetime`Price`Volume,[SYMBOL,DATETIME,DOUBLE,INT]) as Trade

• Insert simulated data to the stream table

As the unit of column 'Datetime' is SECOND and DataFrame in pandas can only use DateTime[64] which corresponds to nanotimestamp data type in DolphinDB, we need to convert the data type of column 'Datetime' before inserting the data to the stream table.

import dolphindb as ddb
import pandas as pd
import numpy as np
csv_file = "trades.csv"
csv_data = pd.read_csv(csv_file, dtype={'Symbol':str} )
csv_df = pd.DataFrame(csv_data)
s = ddb.session();
s.connect("127.0.0.1",8848,"admin","123456")

# Upload DataFrame to DolphinDB and convert the data type of column 'Datetime'

s.upload({"tmpData":csv_df})
s.run("data = select Symbol, datetime(Datetime) as Datetime, Price, Volume from tmpData")
s.run("tableInsert(Trade,data)")

OHLC bars can be calculated in moving windows in real time in DolphinDB. Generally there are the following two scenarios:

• Calculations are conducted after windows end

  • Nonoverlapping windows
  • Partially overlapping windows

• Calculations are conducted continuously within the current window

In this example, the output table of the aggregator is also defined as a stream table. The client can subscribe to the output table through Python API and display the calculation results to the Python terminal.

The following script uses Python API to subscribe to the output table OHLC of the real-time aggregation calculation, and print the result.

from threading import Event
import dolphindb as ddb
import pandas as pd
import numpy as np
s = ddb.session()
# set local port 20001 for subscribed streaming data
s.enableStreaming(20001)
def handler(lst):
    print(lst)
# subscribe to the stream table OHLC (local port 8848)
s.subscribe ("127.0.0.1", 8848, handler, "OHLC")
Event().Wait()

You can also connect to DolphinDB database through a data visualization tool such as Grafana to query the output table and display the results in charts.