DolphinDB开发者评测：用多范式编程重新定义时序数据分析

import dolphindb as ddb
import pandas as pd

# 建立连接
sess = ddb.Session()
sess.connect("localhost", 8848, "admin", "123456")

# 在DolphinDB中创建数据库和表
sess.run("""
    db = database("dfs://demo", VALUE, 2024.01M..2024.12M)
    t = table(1000:0, `date`sym`price`volume,
              [DATE, SYMBOL, DOUBLE, INT])
    db.createPartitionedTable(t, "stock", "date")
""")

# 用pandas DataFrame写入数据
df = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=100),
    'sym': ['AAPL', 'GOOG', 'MSFT'] * 33 + ['AAPL'],
    'price': np.random.uniform(100, 500, 100),
    'volume': np.random.randint(1000, 100000, 100)
})
sess.run("tableInsert{loadTable('dfs://demo', 'stock')}", df)

# 查询数据
result = sess.run("""
    select count(*) as cnt, avg(price) as avg_price,
           sum(volume) as total_vol
    from loadTable("dfs://demo", "stock")
    group by sym
""")
print(result)

-- 标准查询，毫无学习成本
select sym, avg(price) as avg_price, max(volume) as max_vol
from stock
where date between 2024.01.01 and 2024.03.31
group by sym
order by avg_price desc

-- 计算每只股票的日收益率
update stock set ret = ratios(price) - 1.0 context by sym;

-- 每天按收益率降序排名
select date, sym, ret, rank(ret, false) as daily_rank
from stock
where isValid(ret)
context by date;

-- 筛选每天收益率前3的股票
select date, sym, ret
from stock
context by date
having rank(ret, false) < 3;

-- 将股票数据转为透视表：每行一个日期，每列一只股票
exec price from stock pivot by date, sym;

-- asof join：为每条交易找到最近一次报价
select sum(abs(price - (bid+ask)/2.0)*qty) / sum(price*qty) as cost
from aj(trades, quotes, `date`sym`time)
group by sym;

-- window join：查找交易前10毫秒内的报价，计算平均中间价
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)
group by sym;

-- SQL中直接使用上下文中的字典变量
empNames = dict(1..10, `Alice`Bob`Jerry`Jessica`Mike
                `Tim`Henry`Anna`Kevin`Jones)
select empNames[first(id)] as name, avg(wage)
from empWages
group by id;

// for循环：1000万次加法
c = array(DOUBLE, 10000000)
for(i in 0:10000000)
    c[i] = a[i] + b[i]

// 向量化：一行搞定，快100倍以上
c = a + b

// 逐窗口计算移动平均：4039ms
timer moving(avg, a, 60);

// 优化后的mavg函数：13ms
timer mavg(a, 60);

// 自定义函数 + Lambda过滤
def getWeekDays(dates){
    return dates[def(x): weekday(x) between 1:5]
}

// 高阶函数：三行代码计算1000万条数据的相关性矩阵
priceMatrix = pivot(avg, price, time.minute(), syms)
retMatrix = each(ratios, priceMatrix) - 1
corrMatrix = cross(corr, retMatrix, retMatrix)

def generateReport(tbl, colNames, colFormat, filter){
    colCount = colNames.size()
    colDefs = array(ANY, colCount)
    for(i in 0:colCount){
        if(colFormat[i] == "")
            colDefs[i] = sqlCol(colNames[i])
        else
            colDefs[i] = sqlCol(colNames[i], format{,colFormat[i]})
    }
    return sql(colDefs, tbl, filter).eval()
}

// 动态生成报表
generateReport(t, ["id","date","price"],
               ["000","MM/dd/yyyy","00.00"],
               < id < 5 or id > 95 >);

// 自定义分布式线性回归
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])
}

def myOLSFinal(result){
    xtx = result[0]
    xty = result[1]
    return xtx.inv().dot(xty)[0]
}

// 在1000万行分布式数据上运行
myOLSEx(sqlDS(<select * from sample>), `y, `x1`x2);

graph LR
    A[传统方案] --> A1[数据库<br>只做存储]
    A1 --> A2[导出数据到<br>Python/Spark]
    A2 --> A3[训练模型]
    A3 --> A4[部署到<br>推理服务]
    A4 --> A5[对接回<br>数据库]

    B[DolphinDB] --> B1[存储 + 计算<br>+ 分析 + ML]
    B1 --> B2[库内直接完成<br>全链路闭环]

    style A fill:#ffcccc,stroke:#cc3333,color:#333
    style A1 fill:#ffcccc,stroke:#cc3333,color:#333
    style A2 fill:#ffcccc,stroke:#cc3333,color:#333
    style A3 fill:#ffcccc,stroke:#cc3333,color:#333
    style A4 fill:#ffcccc,stroke:#cc3333,color:#333
    style A5 fill:#ffcccc,stroke:#cc3333,color:#333
    style B fill:#d4edda,stroke:#28a745,color:#333
    style B1 fill:#d4edda,stroke:#28a745,color:#333
    style B2 fill:#d4edda,stroke:#28a745,color:#333

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
}