代码拉取完成,页面将自动刷新
# -*- coding: utf-8 -*-
'''
依赖包:
pip install dolphindb
'''
#流创建脚本
streaming_pathway_creating_script= '''
login("admin","123456");
clearAllCache();
undef(all);
go;
//清理流函数
def cleanStreamEngines(engineNames){
for(name in engineNames){
try{
dropStreamEngine(name)
}
catch(ex){
}
}
}
def createEmptyStreamTableAsTargetTable(targetTable,streamSharedName,capacity){
targetSchema=targetTable.schema();
sch =select name, typeString as type from targetSchema.colDefs
colName=sch.name
colType=sch.type
snapshotStreamTemp = streamTable(capacity:0, colName, colType)
enableTableShareAndPersistence(table=snapshotStreamTemp, tableName=streamSharedName, cacheSize=capacity)
return snapshotStreamTemp
}
//因子纵表建表函数
def createStreamTableForDoubleEmaFactor(){
factorSchema = table(
array(SYMBOL, 0) as SecurityID,
array(TIMESTAMP, 0) as TradeTime,
array(SYMBOL, 0) as factorName,
array(DOUBLE, 0) as val)
targetStreamTable=createEmptyStreamTableAsTargetTable(factorSchema,"doubleEmaStream",800000)
return targetStreamTable
}
def loadSnapshotTable(){
dbPath="dfs://snapshot_SH_L2_TSDB"
tableName="snapshot_SH_L2_TSDB"
tableHandle=loadTable(dbPath,tableName)
return tableHandle
}
//定义snapshot数据源的dummy和DS
def getDummyAndDs(){
columnPhrase='SecurityID, TradeTime,"doubleEma" as factorName, LastPx from tableHandle where date(TradeTime)<2020.07.01' //查询各列指令
testString="select top 500 "+columnPhrase
testPhrase=parseExpr(testString)//查询指令
fullString="select "+columnPhrase
fullPhrase=parseExpr(fullString)//查询指令
schemaTable=eval(testPhrase)
DataSource=replayDS(fullPhrase,"TradeTime", "TradeTime")//全量数据源
return schemaTable,DataSource
}
//因子函数定义
def sumDiff(x, y){
return (x-y)\(x+y)
}
@state
def doubleEma(price){
ema_20=ema(price, 20)
ema_40=ema(price, 40)
sumDiff_1000=1000 * sumDiff(ema_20, ema_40)
finalFactor=ema(sumDiff_1000,10) - ema(sumDiff_1000, 20)
return finalFactor
}
//注销之前的流表订阅
try{unsubscribeTable(tableName="doubleEmaStream",actionName="python_subscribe")}catch(ex){}
//注销流表
try{dropStreamTable("doubleEmaStream")}catch(ex){}
resultStreamTable = createStreamTableForDoubleEmaFactor()//创建新流表
tableHandle=loadSnapshotTable()
snapshotDummyTable,inputDS=getDummyAndDs()
//创建流引擎
engineName="doubleEmaFactorDemo"
cleanStreamEngines([engineName])//清理之前的同名引擎
demoEngine = createReactiveStateEngine(name=engineName, metrics=<[TradeTime,factorName,doubleEma(LastPx)]>, dummyTable=snapshotDummyTable, outputTable=resultStreamTable, keyColumn="SecurityID")
'''
#流启动脚本
streaming_start_script= '''
demoJobName="streamingFactorDemoJob"
//提交job供执行
submitJob(demoJobName,"snapshot因子半年",replay,inputDS,demoEngine, `TradeTime, `TradeTime, -1,false, 4)
'''
#以下为python代码,
from threading import Event
from dolphindb import session
DDB_server_port=8851
DDB_datanode_host= "192.168.100.3"
DDB_user_name= "admin"
DDB_password= "123456"
#流表共享名
stream_table_shared_name= "doubleEmaStream"
#流订阅行为名
action_name= "python_subscribe"
#python端回调函数,流表传入数据以后在此下游处理python端的业务
def python_callback_handler(msg):
#显示传入的流数据
print("流表传来数据:")
print(msg)
pass
def main():
current_ddb_session=session()
current_ddb_session.connect(DDB_datanode_host,
DDB_server_port,
DDB_user_name,
DDB_password)
执行结果=current_ddb_session.run(streaming_pathway_creating_script)
print("流创建执行完成",执行结果)
try:
#python端收听端口
python_receiving_port=24555
current_ddb_session.enableStreaming(python_receiving_port)
print(f"DDB将订阅的流发送到客户端{python_receiving_port}端口")
current_ddb_session.subscribe(
host=DDB_datanode_host,
port=DDB_server_port,
handler=python_callback_handler,
tableName=stream_table_shared_name,
actionName=action_name,
offset=0,
resub=False,
filter=None
)
except Exception as current_error:
print(current_error)
脚本result=current_ddb_session.run(streaming_start_script)
#阻塞函数退出,保证DDB会话不析构
Event().wait()
pass
if __name__ == '__main__':
main()
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。