Programming Guide

In DolphinDB, the scripting language is fully integrated with SQL statements.

• SQL query can be assigned directly to a variable or as a function parameter.
• SQL query statements can quote variables or functions. If a SQL query involves a distributed table, the variables and functions in the SQL query can be automatically serialized to the corresponding nodes.
• SQL statements can be dynamically generated script.
• SQL statements can operate not only on a table, but also on other data forms (including scalar, vector, matrix, set and dictionary). Tables can be converted into other data forms.

//generate a table of employee wages
emp_wage = table(take(1..10, 100) as id, take(2017.10M + 1..10, 100).sort() as month, take(5000 5500 6000 6500, 100) as wage)

//compute the average wage for each employee. The employee list is stored in the array 'empIds'.
empIds = 3 4 6 7 9
select avg(wage) from emp_wage where id in empIds group by id
id avg_wage
-- --------
3  5500
4  6000
6  6000
7  5500
9  5500

//display average wage and corresponding employee's name. Get employee name from the dictionary 'empName'.
empName = dict(1..10, `Alice`Bob`Jerry`Jessica`Mike`Tim`Henry`Anna`Kevin`Jones)
select empName[first(id)] as name, avg(wage) from emp_wage where id in empIds group by id
id name    avg_wage
-- ------- --------
3  Jerry   5500
4  Jessica 6000
6  Tim     6000
7  Henry   5500
9  Kevin   5500

In the examples above, SQL queries quote a vector and a dictionary directly. By integrating the programming language and SQL queries, the system uses hash tables to solve a problem that would have required subqueries and table joins. It makes the script more concise and easier to understand. More importantly, it improves the performance.

DolphinDB introduces a SQL keyword exec. Compared with select, the result returned by exec can be a scalar, vector or matrix. The following example uses exec and pivot by to return a matrix.

exec first(wage) from emp_wage pivot by month, id

         1    2    3    4    5    6    7    8    9    10
         ---- ---- ---- ---- ---- ---- ---- ---- ---- ----
2017.11M|5000 5500 6000 6500 5000 5500 6000 6500 5000 5500
2017.12M|6000 6500 5000 5500 6000 6500 5000 5500 6000 6500
2018.01M|5000 5500 6000 6500 5000 5500 6000 6500 5000 5500
2018.02M|6000 6500 5000 5500 6000 6500 5000 5500 6000 6500
2018.03M|5000 5500 6000 6500 5000 5500 6000 6500 5000 5500
2018.04M|6000 6500 5000 5500 6000 6500 5000 5500 6000 6500
2018.05M|5000 5500 6000 6500 5000 5500 6000 6500 5000 5500
2018.06M|6000 6500 5000 5500 6000 6500 5000 5500 6000 6500
2018.07M|5000 5500 6000 6500 5000 5500 6000 6500 5000 5500
2018.08M|6000 6500 5000 5500 6000 6500 5000 5500 6000 6500

Some calculations with panel data return the same number of rows as the input data, such as converting stock prices into stock returns and calculating moving window statistics. For these tasks SQL server and PostgreSQL use window functions. DolphinDB offers context by clause in SQL statements. Compared with window functions, context by clause has the following advantages:

• Can be used with not only select statements but also update statements.
• Window functions can only group existing fields, while context by can also group calculated fields.
• Window functions can only use a limited number of functions, whereas context by clause has no limits on what functions can be used. Besides, context by can use arbitrary expressions, such as a combination of multiple functions.
• context by can be used with having clause to filter rows within each group.

In the following example, we use context by to calculate daily stock returns and ranks.

//calculate daily stock returns assuming data for each stock are already ordered.
update trades set ret = ratios(price) - 1.0 context by sym

//calculate the daily rank(in descending order of stock return)of each stock within each day.
select date, symbol,  ret, rank(ret, false) + 1 as rank from trades where isValid(ret) context by date

//select top 10 stocks each day
select date, symbol, ret from trades where isValid(ret) context by date having rank(ret, false) < 10

A paper 101 Formulaic Alphas introduced 101 quant factors used by WorldQuant, one of Wall Street's top quantitative hedge funds. A quant hedge fund calculated these factors with C#. One of the most complicated factor (No.98) needs hundreds of lines of code in C# and takes about 30 minutes to calculate with data for 3,000 stocks in 10 years. In contrast, the same calcuation in DolphinDB only needs 4 lines of code and takes only 2 seconds, an almost 3 orders of magnitude improvement in performance.

//Implementation of No.98 alpha factor. 'stock' is the input table.
def alpha98(stock){
    t = select code, valueDate, adv5, adv15, open, vwap from stock order by valueDate
    update t set rank_open = rank(open), rank_adv15 = rank(adv15) context by valueDate
    update t set decay7 = mavg(mcorr(vwap, msum(adv5, 26), 5), 1..7), decay8 = mavg(mrank(9 - mimin(mcorr(rank_open, rank_adv15, 21), 9), true, 7), 1..8) context by code
    return select code, valueDate, rank(decay7)-rank(decay8) as A98 from t context by valueDate
}

DolphinDB uses columnar data storage and vector programming, which are very friendly for time-series data analysis.

• DolphinDB supports temporal data types with various precisions up to nanosecond. High-frequency data can be conveniently converted into low-frequency data such as second-level, minute-level and hour-level with SQL statements. It can also be converted into data for intervals with specified length with function bar and SQL clause group by.

• DolphinDB offers convenient time-series functions, such as lead, lag, moving window, cumulative window and so on. More importantly, these functions are all optimized with performance 1 to 2 orders of magnitude better than other systems.

• DolphinDB designs 2 ways of nonsynchronous table joins: asof join and window join.

The following example calculates the average salary of a group of people for 3 months before a certain point in time with function wj. For more details about function wj, please refer to window join.

p = table(1 2 3 as id, 2018.06M 2018.07M 2018.07M as month)
s = table(1 2 1 2 1 2 as id, 2018.04M + 0 0 1 1 2 2 as month, 4500 5000 6000 5000 6000 4500 as wage)
select * from wj(p, s, -3:-1,<avg(wage)>,`id`month)

id month    avg_wage
-- -------- -----------
1  2018.06M 5250
2  2018.07M 4833.333333
3  2018.07M

A typical application of window join in quant finance is to calculate transaction costs with a 'trades' table and a 'quotes' table.

//table 'trades'
sym  date       time         price  qty
---- ---------- ------------ ------ ---
IBM  2018.06.01 10:01:01.005 143.19 100
MSFT 2018.06.01 10:01:04.006 107.94 200
...

//table 'quotes'
sym  date       time         bid    ask    bidSize askSize
---- ---------- ------------ ------ ------ ------- -------
IBM  2018.06.01 10:01:01.006 143.18 143.21 400     200
MSFT 2018.06.01 10:01:04.010 107.92 107.97 800     100
...

dateRange = 2018.05.01 : 2018.08.01

//select the last quote before each transaction with asof join, and use the average of bid and ask as the benchmark for transaction costs.
select sum(abs(price - (bid+ask)/2.0)*qty)/sum(price*qty) as cost from aj(trades, quotes, `date`sym`time) where date between dataRange group by sym

//select all the quotes in the 10 ms before each transaction with window join, and calculate the average mid price as the benchmark for transaction costs.
select sum(abs(price - mid)*qty)/sum(price*qty) as cost from pwj(trades, quotes, -10:0, <avg((bid + ask)/2.0) as mid>,`date`sym`time) where date between dataRange group by sym

DolphinDB SQL offers various other new features. For examples:

• User-defined functions can be used in SQL on the local node or distributed environment without compiling, packaging or deploying.
• SQL is fully integrated with the distributed computing framework in DolphinDB, making in-database analytics more convenient and efficient.
• DolphinDB supports composite column, which can output multiple return values of complex analysis functions into one row of a table.

When using composite columns in SQL statements, the function must output a key-value pair or an array. Otherwise you can convert the result into one of these two types with a user-defined function. For details please refer to User Manual.

factor1=3.2 1.2 5.9 6.9 11.1 9.6 1.4 7.3 2.0 0.1 6.1 2.9 6.3 8.4 5.6
factor2=1.7 1.3 4.2 6.8 9.2 1.3 1.4 7.8 7.9 9.9 9.3 4.6 7.8 2.4 8.7
t=table(take(1 2 3, 15).sort() as id, 1..15 as y, factor1, factor2)

//run `ols` for each id: y = alpha + beta1 * factor1 + beta2 * factor2, and output parameters alpha, beta1 and beta2.
select ols(y, [factor1,factor2], true, 0) as `alpha`beta1`beta2 from t group by id

id alpha     beta1     beta2
-- --------- --------- ---------
1  1.063991  -0.258685 0.732795
2  6.886877  -0.148325 0.303584
3  11.833867 0.272352  -0.065526

//output R2 with a user-defined function.
def myols(y,x){
    r=ols(y,x,true,2)
    return r.Coefficient.beta join r.RegressionStat.statistics[0]
}
select myols(y,[factor1,factor2]) as `alpha`beta1`beta2`R2 from t group by id

id alpha     beta1     beta2     R2
-- --------- --------- --------- --------
1  1.063991  -0.258685 0.732795  0.946056
2  6.886877  -0.148325 0.303584  0.992413
3  11.833867 0.272352  -0.065526 0.144837

A user-defined function can be created with or without function name (usually lamdba function). DolphinDB is different from Python in that only function parameters and local variables can be referred inside a function. This greatly improves the software quality with slight reduction in the flexibility of syntactic sugar.

//define a function to return weekdays
def getWorkDays(dates){
    return dates[def(x):weekday(x) between 1:5]
}

getWorkDays(2018.07.01 2018.08.01 2018.09.01 2018.10.01)

[2018.08.01, 2018.10.01]

In the example above, we defined a lambda function getWorkDays(). The function accepts an array of dates and returns weekdays.

In this section we focus on higher-order functions and partial applications. A higher-order function accepts another function as a parameter. In DolphinDB, higher-order functions are mainly used as template functions. Usually the first parameter is another function. For example, object A has m elements and object B has n elements. To calculate pairwise correlation of the elements in A and B, we need to use function corr on all possible pairs of elements in A and elements in B, and arrange the results in an m*n matrix. In DolphinDB, the aforementioned steps can be simplified with a high-order function cross: cross(corr, A, B). DolphinDB provides many such template functions including all, any, each, loop, eachLeft, eachRight, eachPre, eachPost, accumulate, reduce, groupby, contextby, pivot, cross, moving, rolling, etc.

In the following example, we use 3 higher-order functions to calculate the pairwise correlation with tick-level stock data.

//simulate 100 million data points

n=10000000
syms = rand(`FB`GOOG`MSFT`AMZN`IBM, n)
time = 09:30:00.000 + rand(21600000, n)
price = 500.0 + rand(500.0, n)

//generate a pivot table with `pivot` function
priceMatrix = pivot(avg, price, time.minute(), syms)

//generate stock return per minute for each stock (each column of the matrix) with functions `each` and `ratios`
retMatrix = each(ratios, priceMatrix) - 1

//compute pairwise correlation with funtions `cross` and `corr`
corrMatrix = cross(corr, retMatrix, retMatrix)

     AMZN      FB        GOOG      IBM       MSFT
     --------- --------- --------- --------- ---------
AMZN|1         0.015181  -0.056245 0.005822  0.084104
FB  |0.015181  1         -0.028113 0.034159  -0.117279
GOOG|-0.056245 -0.028113 1         -0.039278 -0.025165
IBM |0.005822  0.034159  -0.039278 1         -0.049922
MSFT|0.084104  -0.117279 -0.025165 -0.049922 1

Partial application fixes part or all of the parameters of a function to produce a new function with fewer parameters. In DolphinDB, function calls use () and partial applications use {}. For example, the ratios function is a partial application with the higher order function eachPre: eachPre{ratio}.

The following 2 lines of script are equivalent:

`retMatrix = each(ratios, priceMatrix) - 1`
retMatrix = each(eachPre{ratio}, priceMatrix) - 1

Partial applications are often used with higher order functions. Higher order functions usually have certain requirements for parameters, and partial applications can ensure that the parameters meet these requirements. For example, to calculate the correlation between vector a and each column in a matrix m, we can use the higher order function each. However, if we use function corr, vector a and matrix m as the parameters of each, the system would try to calculate the correlation between each element of vector a and each column of matrix m, which will throw an exception. We can use partial application to form a new function corr, and then use it as a parameter for the higher order function each together with matrix m. We can also use 'for' statement for this problem, but the script is lengthy and the execution takes longer time.

a = 12 14 18
m = matrix(5 6 7, 1 3 2, 8 7 11)

//calculate correlation with `each` and partial applications
each(corr{a}, m)

//calculate with 'for' statement
cols = m.columns()
c = array(DOUBLE, cols)
for(i in 0:cols)
    c[i] = corr(a, m[i])

Partial application can also be used to keep a function in a state. Usually the result of a function is completely determined by the input parameters. Occasionally, however, we would like a function to be in a state. For example, in streaming, we define a message handler to accept a new message and return a result. For the message handler to return the average of a column for all the messages received, we can use partial application.

def cumulativeAverage(mutable stat, newNum){
    stat[0] = (stat[0] * stat[1] + newNum)/(stat[1] + 1)
    stat[1] += 1
    return stat[0]
}

msgHandler = cumulativeAverage{0.0 0.0}
each(msgHandler, 1 2 3 4 5)

[1,1.5,2,2.5,3]

We can use xdb to create a connection to a remote node. The remote node can be any node running DolphinDB and doesn't need to be in the current cluster. After the connection is created, we can use remoteRun to execute a function that is registered on the remote node or a locally defined function on the remote node.

h = xdb("localhost", 8081)
//run on the remote node
remoteRun(h, "sum(1 3 5 7)");
16

//the script above can be simplified as
h("sum(1 3 5 7)");
16

//at the remote node, run a function registered on the remote node
h("sum", 1 3 5 7);
16

//run a user-defined function on the remote node
def mysum(x) : reduce(+, x)
h(mysum, 1 3 5 7);
16

//create a shared table 'sales' on the remote node (localhost:8081)
h("share table(2018.07.02 2018.07.02 2018.07.03 as date, 1 2 3 as qty, 10 15 7 as price) as sales")
//when we execute a user-defined function on a remote node, its dependent functions will be automatically serialized to the remote node.
defg salesSum(tableName, d): select mysum(price*qty) from objByName(tableName) where date=d
h(salesSum, "sales", 2018.07.02);
40

Another way of remote procedure call is function rpc. Its parameters are the remote node alias, the function to be executed and the function parameters. Both the calling node and the remote node must be located in the same cluster. Otherwise, we need to use function remoteRun. Function rpc doesn't need a new connection. Instead, it reuses existing network connections. This way it saves network resources and the time to create new connections, especially when the cluster has many users. Please note that rpc can only execute one function on a remote node. To execute script with rpc, the script must be wrapped in a user-defined function.

//'sales' is a shared table on a remote node 'nodeB'
rpc("nodeB", salesSum, "sales",2018.07.02);
40

//when using `rpc`, we recommend using partial application to increase readability of the script
rpc("nodeB", salesSum{"sales", 2018.07.02});
40

//create a user on the controller node 'master'
rpc("master", createUser{"jerry", "123456"})

//`rpc` can also call a built-in function or a user-defined function to run on a remote node
rpc("nodeB", reduce{+, 1 2 3 4 5});
15

The most powerful feature of built-in remote call funcions (remoteRun and rpc) is that we can send a local function that calls local user-defined functions to run on a remote node on the fly without compilation or deployment. The system automatically serializes the function definition and the definitions of all dependent functions together with necessary local data to remote nodes. In other systems we can only remote call functions without any user-defined dependent functions.

DolphinDB provides numerous functions for indirect remote procedure calls. For examples, linear regression on distributed database uses rpc and olsEx; function pnodeRun calls a function on multiple remote nodes in parallel and then merges the returned results. pnodeRun is a quite useful tool in cluster management.

//return the most recent 10 running or completed batch jobs on each node
pnodeRun(getRecentJobs{10})

//return the most recent 10 SQL queries on 'nodeA' and 'nodeB'
pnodeRun(getCompletedQueries{10}, `nodeA`nodeB)

//clear caches on all data nodes
pnodeRun(clearAllCache)

DolphinDB provides function mr for distributed computing and imr for iterative distributed computing based on MapReduce. Users only need specify a distributed data source and functions such as map function, reduce function, final function and terminate function.

In the example below, we illustrate distributed median calculation and distributed linear regression.

n=10000000
x1 = pow(rand(1.0,n), 2)
x2 = norm(3.0:1.0, n)
y = 0.5 + 3 * x1 - 0.5*x2 + norm(0.0:1.0, n)
t=table(rand(10, n) as id, y, x1, x2)

login(`admin,"123456")
db = database("dfs://testdb", VALUE, 0..9)
db.createPartitionedTable(t, "sample", "id").append!(t)

Construct a function myOLSEx to conduct a distributed linear regression with user-defined map function myOLSMap, built-in reduce function add, user-defined final function myOLSFinal, and built-in function mr.

//define a map function 'myOLSMap'
def myOLSMap(table, yColName, xColNames){
    x = matrix(take(1.0, table.rows()), table[xColNames])
    xt = x.transpose()
    return xt.dot(x), xt.dot(table[yColName])
}

//define a final function 'myOLSFinal'
def myOLSFinal(result){
    xtx = result[0]
    xty = result[1]
    return xtx.inv().dot(xty)[0]
}

def myOLSEx(ds, yColName, xColNames){
  return mr(ds, myOLSMap{, yColName, xColNames}, +, myOLSFinal)
}

//calculate the linear regression coefficients on distributed data source with function 'myOLSEx'
sample = loadTable("dfs://testdb", "sample")
myOLSEx(sqlDS(<select * from sample>), `y, `x1`x2);
[0.4991, 3.0001, -0.4996]

//get the same result by running the built-in function `ols` on unpartitioned data
ols(y, [x1,x2],true);
[0.4991, 3.0001, -0.4996]

The following example calculates the approximate median value based on a distributed data source. The data is scattered on multiple nodes. First, for each data source, divide the data into buckets and use the map function to count the number of data points in each bucket. Then use the reduce function to calculate the number of data points in each bucket across all data sources. Locate the bucket that contains the median. In the next iteration, the chosen bucket is divided into multiple buckets and the bucket with the median value is located. The iterations is finished when the size of the chosen bucket is no more than the specified precision.

def medMap(data, range, colName): bucketCount(data[colName], double(range), 1024, true)

def medFinal(range, result){
    x= result.cumsum()
    index = x.asof(x[1025]/2.0)
    ranges = range[1] - range[0]
    if(index == -1)
        return (range[0] - ranges*32):range[1]
    else if(index == 1024)
        return range[0]:(range[1] + ranges*32)
    else{
        interval = ranges / 1024.0
        startValue = range[0] + (index - 1) * interval
        return startValue : (startValue + interval)
    }
}

def medEx(ds, colName, range, precision){
    termFunc = def(prev, cur): cur[1] - cur[0] <= precision
    return imr(ds, range, medMap{,,colName}, +, medFinal, termFunc).avg()
}

sample = loadTable("dfs://testdb", "sample")
medEx(sqlDS(<select y from sample>), `y, 0.0 : 1.0, 0.001);
-0.052973

//calculate the median of unpartitioned data with a built-in function `med`
med(y);
-0.052947