python代码
国盛证券qmt mini模式 xtquant
券商万一免五 • 李魔佛 发表了文章 • 0 个评论 • 9688 次浏览 • 2022-07-29 01:00
国盛证券QMT：
【国盛QMT支持 xtquant qmt mini模式】
mini模式可以在外部运行，同时可以下载历史tick数据。xtdata是xtquant库中提供行情相关数据的模块，本模块旨在提供精简直接的数据满足量化交易者的数
据需求，作为python库的形式可以被灵活添加到各种策略脚本中。
主要提供行情数据（历史和实时的K线和分笔）、财务数据、合约基础信息、板块和行业分类信息等通
用的行情数据
可以直接获取level2的数据
使用python代码直接运行，不用在qmt软件里面憋屈地写代码，可直接使用pycharm，vscode编写，且有代码提示，补全，好用多了。
附一个完整的策略例子。
保存为： demo.py
命令行下运行：
python demo.py# 创建策略
#coding=utf-8
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xtquant import StockAccount
from xtquant import xtconstant
class MyXtQuantTraderCallback(XtQuantTraderCallback):
def on_disconnected(self):
"""
连接断开
:return:
"""
print("connection lost")
def on_stock_order(self, order):
"""
委托回报推送
:param order: XtOrder对象
:return:
"""
print("on order callback:")
print(order.stock_code, order.order_status, order.order_sysid)
def on_stock_asset(self, asset):
"""
资金变动推送
:param asset: XtAsset对象
:return:
"""
print("on asset callback")
print(asset.account_id, asset.cash, asset.total_asset)
def on_stock_trade(self, trade):
"""
成交变动推送
:param trade: XtTrade对象
:return:
"""
print("on trade callback")
print(trade.account_id, trade.stock_code, trade.order_id)
def on_stock_position(self, position):
"""
持仓变动推送
:param position: XtPosition对象
:return:
"""
print("on position callback")
print(position.stock_code, position.volume)
def on_order_error(self, order_error):
"""
委托失败推送
:param order_error:XtOrderError 对象
:return:
"""
print("on order_error callback")
print(order_error.order_id, order_error.error_id, order_error.error_msg)
def on_cancel_error(self, cancel_error):
"""
撤单失败推送
:param cancel_error: XtCancelError 对象
:return:
"""
print("on cancel_error callback")
print(cancel_error.order_id, cancel_error.error_id,
cancel_error.error_msg)
def on_order_stock_async_response(self, response):
"""
异步下单回报推送
:param response: XtOrderResponse 对象
:return:
"""
print("on_order_stock_async_response")
print(response.account_id, response.order_id, response.seq)
if __name__ == "__main__":
print("demo test")
# path为mini qmt客户端安装目录下userdata_mini路径
path = 'D:\\迅投极速交易终端 睿智融科版\\userdata_mini'
# session_id为会话编号，策略使用方对于不同的Python策略需要使用不同的会话编号
session_id = 123456
xt_trader = XtQuantTrader(path, session_id)
# 创建资金账号为1000000365的证券账号对象
acc = StockAccount('1000000365')
# 创建交易回调类对象，并声明接收回调
callback = MyXtQuantTraderCallback()
xt_trader.register_callback(callback)
# 启动交易线程
xt_trader.start()
# 建立交易连接，返回0表示连接成功
connect_result = xt_trader.connect()
print(connect_result)
# 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
subscribe_result = xt_trader.subscribe(acc)
print(subscribe_result)
stock_code = '600000.SH'
# 使用指定价下单，接口返回订单编号，后续可以用于撤单操作以及查询委托状态
print("order using the fix price:")
fix_result_order_id = xt_trader.order_stock(acc, stock_code,
xtconstant.STOCK_BUY, 200, xtconstant.FIX_PRICE, 10.5, 'strategy_name',
'remark')
print(fix_result_order_id)
# 使用订单编号撤单
print("cancel order:")
cancel_order_result = xt_trader.cancel_order_stock(acc, fix_result_order_id)
print(cancel_order_result)
# 使用异步下单接口，接口返回下单请求序号seq，seq可以和on_order_stock_async_response
的委托反馈response对应起来
print("order using async api:")
async_seq = xt_trader.order_stock(acc, stock_code, xtconstant.STOCK_BUY,
200, xtconstant.FIX_PRICE, 10.5, 'strategy_name', 'remark')
print(async_seq)
# 查询证券资产
print("query asset:")
asset = xt_trader.query_stock_asset(acc)
if asset:
print("asset:")
print("cash {0}".format(asset.cash))
# 根据订单编号查询委托
print("query order:")
order = xt_trader.query_stock_order(acc, fix_result_order_id)
if order:
print("order:")
print("order {0}".format(order.order_id))
# 查询当日所有的委托
print("query orders:")
orders = xt_trader.query_stock_orders(acc)
print("orders:", len(orders))
if len(orders) != 0:
print("last order:")
print("{0} {1} {2}".format(orders[-1].stock_code,
orders[-1].order_volume, orders[-1].price))
# 查询当日所有的成交
print("query trade:")
trades = xt_trader.query_stock_trades(acc)
print("trades:", len(trades))
if len(trades) != 0:
print("last trade:")
print("{0} {1} {2}".format(trades[-1].stock_code,
trades[-1].traded_volume, trades[-1].traded_price))
# 查询当日所有的持仓
print("query positions:")
positions = xt_trader.query_stock_positions(acc)
print("positions:", len(positions))
if len(positions) != 0:
print("last position:")
print("{0} {1} {2}".format(positions[-1].account_id,
positions[-1].stock_code, positions[-1].volume))
# 根据股票代码查询对应持仓
print("query position:")
position = xt_trader.query_stock_position(acc, stock_code)
if position:
print("position:")
print("{0} {1} {2}".format(position.account_id, position.stock_code,
position.volume))
# 阻塞线程，接收交易推送
xt_trader.run_forever()
开通xtquant的方式可以咨询。
目前开户费率低，门槛低，提供技术支持与交流。
需要的朋友，可以扫码咨询：
备注开户 查看全部
【国盛QMT支持 xtquant qmt mini模式】
mini模式可以在外部运行，同时可以下载历史tick数据。xtdata是xtquant库中提供行情相关数据的模块，本模块旨在提供精简直接的数据满足量化交易者的数
据需求，作为python库的形式可以被灵活添加到各种策略脚本中。
主要提供行情数据（历史和实时的K线和分笔）、财务数据、合约基础信息、板块和行业分类信息等通
用的行情数据
可以直接获取level2的数据
使用python代码直接运行，不用在qmt软件里面憋屈地写代码，可直接使用pycharm，vscode编写，且有代码提示，补全，好用多了。
附一个完整的策略例子。
保存为： demo.py
命令行下运行：
python demo.py# 创建策略
#coding=utf-8
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xtquant import StockAccount
from xtquant import xtconstant
class MyXtQuantTraderCallback(XtQuantTraderCallback):
def on_disconnected(self):
"""
连接断开
:return:
"""
print("connection lost")
def on_stock_order(self, order):
"""
委托回报推送
:param order: XtOrder对象
:return:
"""
print("on order callback:")
print(order.stock_code, order.order_status, order.order_sysid)
def on_stock_asset(self, asset):
"""
资金变动推送
:param asset: XtAsset对象
:return:
"""
print("on asset callback")
print(asset.account_id, asset.cash, asset.total_asset)
def on_stock_trade(self, trade):
"""
成交变动推送
:param trade: XtTrade对象
:return:
"""
print("on trade callback")
print(trade.account_id, trade.stock_code, trade.order_id)
def on_stock_position(self, position):
"""
持仓变动推送
:param position: XtPosition对象
:return:
"""
print("on position callback")
print(position.stock_code, position.volume)
def on_order_error(self, order_error):
"""
委托失败推送
:param order_error:XtOrderError 对象
:return:
"""
print("on order_error callback")
print(order_error.order_id, order_error.error_id, order_error.error_msg)
def on_cancel_error(self, cancel_error):
"""
撤单失败推送
:param cancel_error: XtCancelError 对象
:return:
"""
print("on cancel_error callback")
print(cancel_error.order_id, cancel_error.error_id,
cancel_error.error_msg)
def on_order_stock_async_response(self, response):
"""
异步下单回报推送
:param response: XtOrderResponse 对象
:return:
"""
print("on_order_stock_async_response")
print(response.account_id, response.order_id, response.seq)
if __name__ == "__main__":
print("demo test")
# path为mini qmt客户端安装目录下userdata_mini路径
path = 'D:\\迅投极速交易终端 睿智融科版\\userdata_mini'
# session_id为会话编号，策略使用方对于不同的Python策略需要使用不同的会话编号
session_id = 123456
xt_trader = XtQuantTrader(path, session_id)
# 创建资金账号为1000000365的证券账号对象
acc = StockAccount('1000000365')
# 创建交易回调类对象，并声明接收回调
callback = MyXtQuantTraderCallback()
xt_trader.register_callback(callback)
# 启动交易线程
xt_trader.start()
# 建立交易连接，返回0表示连接成功
connect_result = xt_trader.connect()
print(connect_result)
# 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
subscribe_result = xt_trader.subscribe(acc)
print(subscribe_result)
stock_code = '600000.SH'
# 使用指定价下单，接口返回订单编号，后续可以用于撤单操作以及查询委托状态
print("order using the fix price:")
fix_result_order_id = xt_trader.order_stock(acc, stock_code,
xtconstant.STOCK_BUY, 200, xtconstant.FIX_PRICE, 10.5, 'strategy_name',
'remark')
print(fix_result_order_id)
# 使用订单编号撤单
print("cancel order:")
cancel_order_result = xt_trader.cancel_order_stock(acc, fix_result_order_id)
print(cancel_order_result)
# 使用异步下单接口，接口返回下单请求序号seq，seq可以和on_order_stock_async_response
的委托反馈response对应起来
print("order using async api:")
async_seq = xt_trader.order_stock(acc, stock_code, xtconstant.STOCK_BUY,
200, xtconstant.FIX_PRICE, 10.5, 'strategy_name', 'remark')
print(async_seq)
# 查询证券资产
print("query asset:")
asset = xt_trader.query_stock_asset(acc)
if asset:
print("asset:")
print("cash {0}".format(asset.cash))
# 根据订单编号查询委托
print("query order:")
order = xt_trader.query_stock_order(acc, fix_result_order_id)
if order:
print("order:")
print("order {0}".format(order.order_id))
# 查询当日所有的委托
print("query orders:")
orders = xt_trader.query_stock_orders(acc)
print("orders:", len(orders))
if len(orders) != 0:
print("last order:")
print("{0} {1} {2}".format(orders[-1].stock_code,
orders[-1].order_volume, orders[-1].price))
# 查询当日所有的成交
print("query trade:")
trades = xt_trader.query_stock_trades(acc)
print("trades:", len(trades))
if len(trades) != 0:
print("last trade:")
print("{0} {1} {2}".format(trades[-1].stock_code,
trades[-1].traded_volume, trades[-1].traded_price))
# 查询当日所有的持仓
print("query positions:")
positions = xt_trader.query_stock_positions(acc)
print("positions:", len(positions))
if len(positions) != 0:
print("last position:")
print("{0} {1} {2}".format(positions[-1].account_id,
positions[-1].stock_code, positions[-1].volume))
# 根据股票代码查询对应持仓
print("query position:")
position = xt_trader.query_stock_position(acc, stock_code)
if position:
print("position:")
print("{0} {1} {2}".format(position.account_id, position.stock_code,
position.volume))
# 阻塞线程，接收交易推送
xt_trader.run_forever()
开通xtquant的方式可以咨询。
目前开户费率低，门槛低，提供技术支持与交流。
需要的朋友，可以扫码咨询：
备注开户 查看全部
国盛证券QMT：
【国盛QMT支持 xtquant qmt mini模式】
mini模式可以在外部运行，同时可以下载历史tick数据。
可以直接获取level2的数据
使用python代码直接运行，不用在qmt软件里面憋屈地写代码，可直接使用pycharm，vscode编写，且有代码提示，补全，好用多了。
附一个完整的策略例子。
保存为： demo.py
命令行下运行：
python demo.py
开通xtquant的方式可以咨询。
目前开户费率低，门槛低，提供技术支持与交流。
需要的朋友，可以扫码咨询：
备注开户
【国盛QMT支持 xtquant qmt mini模式】
mini模式可以在外部运行，同时可以下载历史tick数据。
xtdata是xtquant库中提供行情相关数据的模块，本模块旨在提供精简直接的数据满足量化交易者的数
据需求，作为python库的形式可以被灵活添加到各种策略脚本中。
主要提供行情数据（历史和实时的K线和分笔）、财务数据、合约基础信息、板块和行业分类信息等通
用的行情数据
可以直接获取level2的数据
使用python代码直接运行，不用在qmt软件里面憋屈地写代码，可直接使用pycharm，vscode编写，且有代码提示，补全，好用多了。
附一个完整的策略例子。
保存为： demo.py
命令行下运行：
python demo.py
# 创建策略
#coding=utf-8
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xtquant import StockAccount
from xtquant import xtconstant
class MyXtQuantTraderCallback(XtQuantTraderCallback):
def on_disconnected(self):
"""
连接断开
:return:
"""
print("connection lost")
def on_stock_order(self, order):
"""
委托回报推送
:param order: XtOrder对象
:return:
"""
print("on order callback:")
print(order.stock_code, order.order_status, order.order_sysid)
def on_stock_asset(self, asset):
"""
资金变动推送
:param asset: XtAsset对象
:return:
"""
print("on asset callback")
print(asset.account_id, asset.cash, asset.total_asset)
def on_stock_trade(self, trade):
"""
成交变动推送
:param trade: XtTrade对象
:return:
"""
print("on trade callback")
print(trade.account_id, trade.stock_code, trade.order_id)
def on_stock_position(self, position):
"""
持仓变动推送
:param position: XtPosition对象
:return:
"""
print("on position callback")
print(position.stock_code, position.volume)
def on_order_error(self, order_error):
"""
委托失败推送
:param order_error:XtOrderError 对象
:return:
"""
print("on order_error callback")
print(order_error.order_id, order_error.error_id, order_error.error_msg)
def on_cancel_error(self, cancel_error):
"""
撤单失败推送
:param cancel_error: XtCancelError 对象
:return:
"""
print("on cancel_error callback")
print(cancel_error.order_id, cancel_error.error_id,
cancel_error.error_msg)
def on_order_stock_async_response(self, response):
"""
异步下单回报推送
:param response: XtOrderResponse 对象
:return:
"""
print("on_order_stock_async_response")
print(response.account_id, response.order_id, response.seq)
if __name__ == "__main__":
print("demo test")
# path为mini qmt客户端安装目录下userdata_mini路径
path = 'D:\\迅投极速交易终端 睿智融科版\\userdata_mini'
# session_id为会话编号，策略使用方对于不同的Python策略需要使用不同的会话编号
session_id = 123456
xt_trader = XtQuantTrader(path, session_id)
# 创建资金账号为1000000365的证券账号对象
acc = StockAccount('1000000365')
# 创建交易回调类对象，并声明接收回调
callback = MyXtQuantTraderCallback()
xt_trader.register_callback(callback)
# 启动交易线程
xt_trader.start()
# 建立交易连接，返回0表示连接成功
connect_result = xt_trader.connect()
print(connect_result)
# 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
subscribe_result = xt_trader.subscribe(acc)
print(subscribe_result)
stock_code = '600000.SH'
# 使用指定价下单，接口返回订单编号，后续可以用于撤单操作以及查询委托状态
print("order using the fix price:")
fix_result_order_id = xt_trader.order_stock(acc, stock_code,
xtconstant.STOCK_BUY, 200, xtconstant.FIX_PRICE, 10.5, 'strategy_name',
'remark')
print(fix_result_order_id)
# 使用订单编号撤单
print("cancel order:")
cancel_order_result = xt_trader.cancel_order_stock(acc, fix_result_order_id)
print(cancel_order_result)
# 使用异步下单接口，接口返回下单请求序号seq，seq可以和on_order_stock_async_response
的委托反馈response对应起来
print("order using async api:")
async_seq = xt_trader.order_stock(acc, stock_code, xtconstant.STOCK_BUY,
200, xtconstant.FIX_PRICE, 10.5, 'strategy_name', 'remark')
print(async_seq)
# 查询证券资产
print("query asset:")
asset = xt_trader.query_stock_asset(acc)
if asset:
print("asset:")
print("cash {0}".format(asset.cash))
# 根据订单编号查询委托
print("query order:")
order = xt_trader.query_stock_order(acc, fix_result_order_id)
if order:
print("order:")
print("order {0}".format(order.order_id))
# 查询当日所有的委托
print("query orders:")
orders = xt_trader.query_stock_orders(acc)
print("orders:", len(orders))
if len(orders) != 0:
print("last order:")
print("{0} {1} {2}".format(orders[-1].stock_code,
orders[-1].order_volume, orders[-1].price))
# 查询当日所有的成交
print("query trade:")
trades = xt_trader.query_stock_trades(acc)
print("trades:", len(trades))
if len(trades) != 0:
print("last trade:")
print("{0} {1} {2}".format(trades[-1].stock_code,
trades[-1].traded_volume, trades[-1].traded_price))
# 查询当日所有的持仓
print("query positions:")
positions = xt_trader.query_stock_positions(acc)
print("positions:", len(positions))
if len(positions) != 0:
print("last position:")
print("{0} {1} {2}".format(positions[-1].account_id,
positions[-1].stock_code, positions[-1].volume))
# 根据股票代码查询对应持仓
print("query position:")
position = xt_trader.query_stock_position(acc, stock_code)
if position:
print("position:")
print("{0} {1} {2}".format(position.account_id, position.stock_code,
position.volume))
# 阻塞线程，接收交易推送
xt_trader.run_forever()
开通xtquant的方式可以咨询。
目前开户费率低，门槛低，提供技术支持与交流。
需要的朋友，可以扫码咨询：
备注开户
python sqlite3 多线程 批量写入 【代码】
python • 李魔佛 发表了文章 • 0 个评论 • 3320 次浏览 • 2022-07-09 18:55
1. 随机生成一个数组数据
2. 在多线程里面批量插入数据
几个关注点：
sqlite3.connect(_type, check_same_thread=False) 要设置为False
批量写的时候，记得要加锁
import datetime
import random
import sqlite3
import threading
import logging as log
import time
lock = threading.Lock()
class SQLiteDBCls:
def __init__(self, cache=True):
_type = ":memory:"
self.db = sqlite3.connect(_type, check_same_thread=False)
self.table_name = 'tick_data'
def create_index(self):
cmd = 'CREATE INDEX code_ix ON {} (current)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def create_table(self):
# cursor = self.db.cursor()
cmd = 'create table if not exists {} (id INTEGER PRIMARY KEY AUTOINCREMENT,code text,open double,current time)'.format(
self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def add(self, code, price, t):
cmd = 'insert into {} (code,open,current) values (?,?,?);'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd, (code, price, t))
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def batch_add(self, data):
# 批量加入
print('===========',threading.current_thread().getName())
# log.info(threading.current_thread().getName())
cmd = 'insert into {} (code,open,current) values (?,?,?)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.executemany(cmd, data)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def result(self):
cmd = 'select count(*) from `{}`'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
return cursor.fetchone()
def data_gen():
minute = 6000
code = ['123011.SS','110010.SS','112111.SS']
for i in range(minute):
current = (datetime.datetime.now()+datetime.timedelta(minutes=i)).strftime('%H:%M:%D')
data_list =
for c in code:
price = 5+random.random()+120
data = (c,price,current)
data_list.append(data)
yield data_list
# time.sleep(0.5)
app = SQLiteDBCls(cache=True)
app.create_table()
app.create_index()
def data_validation():
print(app.result())
app.sync_up()
def multithread_mode():
total_count = 0
thread_list =
for d in data_gen():
print(d)
total_count+=len(d)
# app.batch_add(d)
t=threading.Thread(target=app.batch_add,args=(d,))
thread_list.append(t)
for t in thread_list:
t.start()
for t in thread_list:
t.join()
print(total_count)
if __name__=='__main__':
multithread_mode()
data_validation()
假如不加锁会出错：
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 910, in run
self._target(*self._args, **self._kwargs)
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 77, in batch_add
self.db.commit()
Exception in thread Thread-3824:
Exception in thread Thread-3826:
Traceback (most recent call last):
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 973, in _bootstrap_inner
sqlite3.OperationalError: cannot commit - no transaction is activeTraceback (most recent call last):
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 72, in batch_add
cursor.executemany(cmd, data)
sqlite3.InterfaceError: Error binding parameter 0 - probably unsupported type.
查看全部
2. 在多线程里面批量插入数据
几个关注点：
sqlite3.connect(_type, check_same_thread=False) 要设置为False
批量写的时候，记得要加锁
import datetime
import random
import sqlite3
import threading
import logging as log
import time
lock = threading.Lock()
class SQLiteDBCls:
def __init__(self, cache=True):
_type = ":memory:"
self.db = sqlite3.connect(_type, check_same_thread=False)
self.table_name = 'tick_data'
def create_index(self):
cmd = 'CREATE INDEX code_ix ON {} (current)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def create_table(self):
# cursor = self.db.cursor()
cmd = 'create table if not exists {} (id INTEGER PRIMARY KEY AUTOINCREMENT,code text,open double,current time)'.format(
self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def add(self, code, price, t):
cmd = 'insert into {} (code,open,current) values (?,?,?);'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd, (code, price, t))
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def batch_add(self, data):
# 批量加入
print('===========',threading.current_thread().getName())
# log.info(threading.current_thread().getName())
cmd = 'insert into {} (code,open,current) values (?,?,?)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.executemany(cmd, data)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def result(self):
cmd = 'select count(*) from `{}`'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
return cursor.fetchone()
def data_gen():
minute = 6000
code = ['123011.SS','110010.SS','112111.SS']
for i in range(minute):
current = (datetime.datetime.now()+datetime.timedelta(minutes=i)).strftime('%H:%M:%D')
data_list =
for c in code:
price = 5+random.random()+120
data = (c,price,current)
data_list.append(data)
yield data_list
# time.sleep(0.5)
app = SQLiteDBCls(cache=True)
app.create_table()
app.create_index()
def data_validation():
print(app.result())
app.sync_up()
def multithread_mode():
total_count = 0
thread_list =
for d in data_gen():
print(d)
total_count+=len(d)
# app.batch_add(d)
t=threading.Thread(target=app.batch_add,args=(d,))
thread_list.append(t)
for t in thread_list:
t.start()
for t in thread_list:
t.join()
print(total_count)
if __name__=='__main__':
multithread_mode()
data_validation()
假如不加锁会出错：
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 910, in run
self._target(*self._args, **self._kwargs)
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 77, in batch_add
self.db.commit()
Exception in thread Thread-3824:
Exception in thread Thread-3826:
Traceback (most recent call last):
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 973, in _bootstrap_inner
sqlite3.OperationalError: cannot commit - no transaction is activeTraceback (most recent call last):
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 72, in batch_add
cursor.executemany(cmd, data)
sqlite3.InterfaceError: Error binding parameter 0 - probably unsupported type.
查看全部
1. 随机生成一个数组数据
2. 在多线程里面批量插入数据
几个关注点：
sqlite3.connect(_type, check_same_thread=False) 要设置为False
批量写的时候，记得要加锁
假如不加锁会出错：
2. 在多线程里面批量插入数据
几个关注点：
sqlite3.connect(_type, check_same_thread=False) 要设置为False
批量写的时候，记得要加锁
import datetime
import random
import sqlite3
import threading
import logging as log
import time
lock = threading.Lock()
class SQLiteDBCls:
def __init__(self, cache=True):
_type = ":memory:"
self.db = sqlite3.connect(_type, check_same_thread=False)
self.table_name = 'tick_data'
def create_index(self):
cmd = 'CREATE INDEX code_ix ON {} (current)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def create_table(self):
# cursor = self.db.cursor()
cmd = 'create table if not exists {} (id INTEGER PRIMARY KEY AUTOINCREMENT,code text,open double,current time)'.format(
self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def add(self, code, price, t):
cmd = 'insert into {} (code,open,current) values (?,?,?);'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd, (code, price, t))
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def batch_add(self, data):
# 批量加入
print('===========',threading.current_thread().getName())
# log.info(threading.current_thread().getName())
cmd = 'insert into {} (code,open,current) values (?,?,?)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.executemany(cmd, data)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def result(self):
cmd = 'select count(*) from `{}`'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
return cursor.fetchone()
def data_gen():
minute = 6000
code = ['123011.SS','110010.SS','112111.SS']
for i in range(minute):
current = (datetime.datetime.now()+datetime.timedelta(minutes=i)).strftime('%H:%M:%D')
data_list =
for c in code:
price = 5+random.random()+120
data = (c,price,current)
data_list.append(data)
yield data_list
# time.sleep(0.5)
app = SQLiteDBCls(cache=True)
app.create_table()
app.create_index()
def data_validation():
print(app.result())
app.sync_up()
def multithread_mode():
total_count = 0
thread_list =
for d in data_gen():
print(d)
total_count+=len(d)
# app.batch_add(d)
t=threading.Thread(target=app.batch_add,args=(d,))
thread_list.append(t)
for t in thread_list:
t.start()
for t in thread_list:
t.join()
print(total_count)
if __name__=='__main__':
multithread_mode()
data_validation()
假如不加锁会出错：
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 910, in run
self._target(*self._args, **self._kwargs)
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 77, in batch_add
self.db.commit()
Exception in thread Thread-3824:
Exception in thread Thread-3826:
Traceback (most recent call last):
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 973, in _bootstrap_inner
sqlite3.OperationalError: cannot commit - no transaction is activeTraceback (most recent call last):
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 72, in batch_add
cursor.executemany(cmd, data)
sqlite3.InterfaceError: Error binding parameter 0 - probably unsupported type.
国盛证券qmt mini模式 xtquant
券商万一免五 • 李魔佛 发表了文章 • 0 个评论 • 9688 次浏览 • 2022-07-29 01:00
国盛证券QMT：
【国盛QMT支持 xtquant qmt mini模式】
mini模式可以在外部运行，同时可以下载历史tick数据。xtdata是xtquant库中提供行情相关数据的模块，本模块旨在提供精简直接的数据满足量化交易者的数
据需求，作为python库的形式可以被灵活添加到各种策略脚本中。
主要提供行情数据（历史和实时的K线和分笔）、财务数据、合约基础信息、板块和行业分类信息等通
用的行情数据
可以直接获取level2的数据
使用python代码直接运行，不用在qmt软件里面憋屈地写代码，可直接使用pycharm，vscode编写，且有代码提示，补全，好用多了。
附一个完整的策略例子。
保存为： demo.py
命令行下运行：
python demo.py# 创建策略
#coding=utf-8
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xtquant import StockAccount
from xtquant import xtconstant
class MyXtQuantTraderCallback(XtQuantTraderCallback):
def on_disconnected(self):
"""
连接断开
:return:
"""
print("connection lost")
def on_stock_order(self, order):
"""
委托回报推送
:param order: XtOrder对象
:return:
"""
print("on order callback:")
print(order.stock_code, order.order_status, order.order_sysid)
def on_stock_asset(self, asset):
"""
资金变动推送
:param asset: XtAsset对象
:return:
"""
print("on asset callback")
print(asset.account_id, asset.cash, asset.total_asset)
def on_stock_trade(self, trade):
"""
成交变动推送
:param trade: XtTrade对象
:return:
"""
print("on trade callback")
print(trade.account_id, trade.stock_code, trade.order_id)
def on_stock_position(self, position):
"""
持仓变动推送
:param position: XtPosition对象
:return:
"""
print("on position callback")
print(position.stock_code, position.volume)
def on_order_error(self, order_error):
"""
委托失败推送
:param order_error:XtOrderError 对象
:return:
"""
print("on order_error callback")
print(order_error.order_id, order_error.error_id, order_error.error_msg)
def on_cancel_error(self, cancel_error):
"""
撤单失败推送
:param cancel_error: XtCancelError 对象
:return:
"""
print("on cancel_error callback")
print(cancel_error.order_id, cancel_error.error_id,
cancel_error.error_msg)
def on_order_stock_async_response(self, response):
"""
异步下单回报推送
:param response: XtOrderResponse 对象
:return:
"""
print("on_order_stock_async_response")
print(response.account_id, response.order_id, response.seq)
if __name__ == "__main__":
print("demo test")
# path为mini qmt客户端安装目录下userdata_mini路径
path = 'D:\\迅投极速交易终端 睿智融科版\\userdata_mini'
# session_id为会话编号，策略使用方对于不同的Python策略需要使用不同的会话编号
session_id = 123456
xt_trader = XtQuantTrader(path, session_id)
# 创建资金账号为1000000365的证券账号对象
acc = StockAccount('1000000365')
# 创建交易回调类对象，并声明接收回调
callback = MyXtQuantTraderCallback()
xt_trader.register_callback(callback)
# 启动交易线程
xt_trader.start()
# 建立交易连接，返回0表示连接成功
connect_result = xt_trader.connect()
print(connect_result)
# 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
subscribe_result = xt_trader.subscribe(acc)
print(subscribe_result)
stock_code = '600000.SH'
# 使用指定价下单，接口返回订单编号，后续可以用于撤单操作以及查询委托状态
print("order using the fix price:")
fix_result_order_id = xt_trader.order_stock(acc, stock_code,
xtconstant.STOCK_BUY, 200, xtconstant.FIX_PRICE, 10.5, 'strategy_name',
'remark')
print(fix_result_order_id)
# 使用订单编号撤单
print("cancel order:")
cancel_order_result = xt_trader.cancel_order_stock(acc, fix_result_order_id)
print(cancel_order_result)
# 使用异步下单接口，接口返回下单请求序号seq，seq可以和on_order_stock_async_response
的委托反馈response对应起来
print("order using async api:")
async_seq = xt_trader.order_stock(acc, stock_code, xtconstant.STOCK_BUY,
200, xtconstant.FIX_PRICE, 10.5, 'strategy_name', 'remark')
print(async_seq)
# 查询证券资产
print("query asset:")
asset = xt_trader.query_stock_asset(acc)
if asset:
print("asset:")
print("cash {0}".format(asset.cash))
# 根据订单编号查询委托
print("query order:")
order = xt_trader.query_stock_order(acc, fix_result_order_id)
if order:
print("order:")
print("order {0}".format(order.order_id))
# 查询当日所有的委托
print("query orders:")
orders = xt_trader.query_stock_orders(acc)
print("orders:", len(orders))
if len(orders) != 0:
print("last order:")
print("{0} {1} {2}".format(orders[-1].stock_code,
orders[-1].order_volume, orders[-1].price))
# 查询当日所有的成交
print("query trade:")
trades = xt_trader.query_stock_trades(acc)
print("trades:", len(trades))
if len(trades) != 0:
print("last trade:")
print("{0} {1} {2}".format(trades[-1].stock_code,
trades[-1].traded_volume, trades[-1].traded_price))
# 查询当日所有的持仓
print("query positions:")
positions = xt_trader.query_stock_positions(acc)
print("positions:", len(positions))
if len(positions) != 0:
print("last position:")
print("{0} {1} {2}".format(positions[-1].account_id,
positions[-1].stock_code, positions[-1].volume))
# 根据股票代码查询对应持仓
print("query position:")
position = xt_trader.query_stock_position(acc, stock_code)
if position:
print("position:")
print("{0} {1} {2}".format(position.account_id, position.stock_code,
position.volume))
# 阻塞线程，接收交易推送
xt_trader.run_forever()
开通xtquant的方式可以咨询。
目前开户费率低，门槛低，提供技术支持与交流。
需要的朋友，可以扫码咨询：
备注开户 查看全部
【国盛QMT支持 xtquant qmt mini模式】
mini模式可以在外部运行，同时可以下载历史tick数据。xtdata是xtquant库中提供行情相关数据的模块，本模块旨在提供精简直接的数据满足量化交易者的数
据需求，作为python库的形式可以被灵活添加到各种策略脚本中。
主要提供行情数据（历史和实时的K线和分笔）、财务数据、合约基础信息、板块和行业分类信息等通
用的行情数据
可以直接获取level2的数据
使用python代码直接运行，不用在qmt软件里面憋屈地写代码，可直接使用pycharm，vscode编写，且有代码提示，补全，好用多了。
附一个完整的策略例子。
保存为： demo.py
命令行下运行：
python demo.py# 创建策略
#coding=utf-8
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xtquant import StockAccount
from xtquant import xtconstant
class MyXtQuantTraderCallback(XtQuantTraderCallback):
def on_disconnected(self):
"""
连接断开
:return:
"""
print("connection lost")
def on_stock_order(self, order):
"""
委托回报推送
:param order: XtOrder对象
:return:
"""
print("on order callback:")
print(order.stock_code, order.order_status, order.order_sysid)
def on_stock_asset(self, asset):
"""
资金变动推送
:param asset: XtAsset对象
:return:
"""
print("on asset callback")
print(asset.account_id, asset.cash, asset.total_asset)
def on_stock_trade(self, trade):
"""
成交变动推送
:param trade: XtTrade对象
:return:
"""
print("on trade callback")
print(trade.account_id, trade.stock_code, trade.order_id)
def on_stock_position(self, position):
"""
持仓变动推送
:param position: XtPosition对象
:return:
"""
print("on position callback")
print(position.stock_code, position.volume)
def on_order_error(self, order_error):
"""
委托失败推送
:param order_error:XtOrderError 对象
:return:
"""
print("on order_error callback")
print(order_error.order_id, order_error.error_id, order_error.error_msg)
def on_cancel_error(self, cancel_error):
"""
撤单失败推送
:param cancel_error: XtCancelError 对象
:return:
"""
print("on cancel_error callback")
print(cancel_error.order_id, cancel_error.error_id,
cancel_error.error_msg)
def on_order_stock_async_response(self, response):
"""
异步下单回报推送
:param response: XtOrderResponse 对象
:return:
"""
print("on_order_stock_async_response")
print(response.account_id, response.order_id, response.seq)
if __name__ == "__main__":
print("demo test")
# path为mini qmt客户端安装目录下userdata_mini路径
path = 'D:\\迅投极速交易终端 睿智融科版\\userdata_mini'
# session_id为会话编号，策略使用方对于不同的Python策略需要使用不同的会话编号
session_id = 123456
xt_trader = XtQuantTrader(path, session_id)
# 创建资金账号为1000000365的证券账号对象
acc = StockAccount('1000000365')
# 创建交易回调类对象，并声明接收回调
callback = MyXtQuantTraderCallback()
xt_trader.register_callback(callback)
# 启动交易线程
xt_trader.start()
# 建立交易连接，返回0表示连接成功
connect_result = xt_trader.connect()
print(connect_result)
# 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
subscribe_result = xt_trader.subscribe(acc)
print(subscribe_result)
stock_code = '600000.SH'
# 使用指定价下单，接口返回订单编号，后续可以用于撤单操作以及查询委托状态
print("order using the fix price:")
fix_result_order_id = xt_trader.order_stock(acc, stock_code,
xtconstant.STOCK_BUY, 200, xtconstant.FIX_PRICE, 10.5, 'strategy_name',
'remark')
print(fix_result_order_id)
# 使用订单编号撤单
print("cancel order:")
cancel_order_result = xt_trader.cancel_order_stock(acc, fix_result_order_id)
print(cancel_order_result)
# 使用异步下单接口，接口返回下单请求序号seq，seq可以和on_order_stock_async_response
的委托反馈response对应起来
print("order using async api:")
async_seq = xt_trader.order_stock(acc, stock_code, xtconstant.STOCK_BUY,
200, xtconstant.FIX_PRICE, 10.5, 'strategy_name', 'remark')
print(async_seq)
# 查询证券资产
print("query asset:")
asset = xt_trader.query_stock_asset(acc)
if asset:
print("asset:")
print("cash {0}".format(asset.cash))
# 根据订单编号查询委托
print("query order:")
order = xt_trader.query_stock_order(acc, fix_result_order_id)
if order:
print("order:")
print("order {0}".format(order.order_id))
# 查询当日所有的委托
print("query orders:")
orders = xt_trader.query_stock_orders(acc)
print("orders:", len(orders))
if len(orders) != 0:
print("last order:")
print("{0} {1} {2}".format(orders[-1].stock_code,
orders[-1].order_volume, orders[-1].price))
# 查询当日所有的成交
print("query trade:")
trades = xt_trader.query_stock_trades(acc)
print("trades:", len(trades))
if len(trades) != 0:
print("last trade:")
print("{0} {1} {2}".format(trades[-1].stock_code,
trades[-1].traded_volume, trades[-1].traded_price))
# 查询当日所有的持仓
print("query positions:")
positions = xt_trader.query_stock_positions(acc)
print("positions:", len(positions))
if len(positions) != 0:
print("last position:")
print("{0} {1} {2}".format(positions[-1].account_id,
positions[-1].stock_code, positions[-1].volume))
# 根据股票代码查询对应持仓
print("query position:")
position = xt_trader.query_stock_position(acc, stock_code)
if position:
print("position:")
print("{0} {1} {2}".format(position.account_id, position.stock_code,
position.volume))
# 阻塞线程，接收交易推送
xt_trader.run_forever()
开通xtquant的方式可以咨询。
目前开户费率低，门槛低，提供技术支持与交流。
需要的朋友，可以扫码咨询：
备注开户 查看全部
国盛证券QMT：
【国盛QMT支持 xtquant qmt mini模式】
mini模式可以在外部运行，同时可以下载历史tick数据。
可以直接获取level2的数据
使用python代码直接运行，不用在qmt软件里面憋屈地写代码，可直接使用pycharm，vscode编写，且有代码提示，补全，好用多了。
附一个完整的策略例子。
保存为： demo.py
命令行下运行：
python demo.py
开通xtquant的方式可以咨询。
目前开户费率低，门槛低，提供技术支持与交流。
需要的朋友，可以扫码咨询：
备注开户
【国盛QMT支持 xtquant qmt mini模式】
mini模式可以在外部运行，同时可以下载历史tick数据。
xtdata是xtquant库中提供行情相关数据的模块，本模块旨在提供精简直接的数据满足量化交易者的数
据需求，作为python库的形式可以被灵活添加到各种策略脚本中。
主要提供行情数据（历史和实时的K线和分笔）、财务数据、合约基础信息、板块和行业分类信息等通
用的行情数据
可以直接获取level2的数据
使用python代码直接运行，不用在qmt软件里面憋屈地写代码，可直接使用pycharm，vscode编写，且有代码提示，补全，好用多了。
附一个完整的策略例子。
保存为： demo.py
命令行下运行：
python demo.py
# 创建策略
#coding=utf-8
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xtquant import StockAccount
from xtquant import xtconstant
class MyXtQuantTraderCallback(XtQuantTraderCallback):
def on_disconnected(self):
"""
连接断开
:return:
"""
print("connection lost")
def on_stock_order(self, order):
"""
委托回报推送
:param order: XtOrder对象
:return:
"""
print("on order callback:")
print(order.stock_code, order.order_status, order.order_sysid)
def on_stock_asset(self, asset):
"""
资金变动推送
:param asset: XtAsset对象
:return:
"""
print("on asset callback")
print(asset.account_id, asset.cash, asset.total_asset)
def on_stock_trade(self, trade):
"""
成交变动推送
:param trade: XtTrade对象
:return:
"""
print("on trade callback")
print(trade.account_id, trade.stock_code, trade.order_id)
def on_stock_position(self, position):
"""
持仓变动推送
:param position: XtPosition对象
:return:
"""
print("on position callback")
print(position.stock_code, position.volume)
def on_order_error(self, order_error):
"""
委托失败推送
:param order_error:XtOrderError 对象
:return:
"""
print("on order_error callback")
print(order_error.order_id, order_error.error_id, order_error.error_msg)
def on_cancel_error(self, cancel_error):
"""
撤单失败推送
:param cancel_error: XtCancelError 对象
:return:
"""
print("on cancel_error callback")
print(cancel_error.order_id, cancel_error.error_id,
cancel_error.error_msg)
def on_order_stock_async_response(self, response):
"""
异步下单回报推送
:param response: XtOrderResponse 对象
:return:
"""
print("on_order_stock_async_response")
print(response.account_id, response.order_id, response.seq)
if __name__ == "__main__":
print("demo test")
# path为mini qmt客户端安装目录下userdata_mini路径
path = 'D:\\迅投极速交易终端 睿智融科版\\userdata_mini'
# session_id为会话编号，策略使用方对于不同的Python策略需要使用不同的会话编号
session_id = 123456
xt_trader = XtQuantTrader(path, session_id)
# 创建资金账号为1000000365的证券账号对象
acc = StockAccount('1000000365')
# 创建交易回调类对象，并声明接收回调
callback = MyXtQuantTraderCallback()
xt_trader.register_callback(callback)
# 启动交易线程
xt_trader.start()
# 建立交易连接，返回0表示连接成功
connect_result = xt_trader.connect()
print(connect_result)
# 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
subscribe_result = xt_trader.subscribe(acc)
print(subscribe_result)
stock_code = '600000.SH'
# 使用指定价下单，接口返回订单编号，后续可以用于撤单操作以及查询委托状态
print("order using the fix price:")
fix_result_order_id = xt_trader.order_stock(acc, stock_code,
xtconstant.STOCK_BUY, 200, xtconstant.FIX_PRICE, 10.5, 'strategy_name',
'remark')
print(fix_result_order_id)
# 使用订单编号撤单
print("cancel order:")
cancel_order_result = xt_trader.cancel_order_stock(acc, fix_result_order_id)
print(cancel_order_result)
# 使用异步下单接口，接口返回下单请求序号seq，seq可以和on_order_stock_async_response
的委托反馈response对应起来
print("order using async api:")
async_seq = xt_trader.order_stock(acc, stock_code, xtconstant.STOCK_BUY,
200, xtconstant.FIX_PRICE, 10.5, 'strategy_name', 'remark')
print(async_seq)
# 查询证券资产
print("query asset:")
asset = xt_trader.query_stock_asset(acc)
if asset:
print("asset:")
print("cash {0}".format(asset.cash))
# 根据订单编号查询委托
print("query order:")
order = xt_trader.query_stock_order(acc, fix_result_order_id)
if order:
print("order:")
print("order {0}".format(order.order_id))
# 查询当日所有的委托
print("query orders:")
orders = xt_trader.query_stock_orders(acc)
print("orders:", len(orders))
if len(orders) != 0:
print("last order:")
print("{0} {1} {2}".format(orders[-1].stock_code,
orders[-1].order_volume, orders[-1].price))
# 查询当日所有的成交
print("query trade:")
trades = xt_trader.query_stock_trades(acc)
print("trades:", len(trades))
if len(trades) != 0:
print("last trade:")
print("{0} {1} {2}".format(trades[-1].stock_code,
trades[-1].traded_volume, trades[-1].traded_price))
# 查询当日所有的持仓
print("query positions:")
positions = xt_trader.query_stock_positions(acc)
print("positions:", len(positions))
if len(positions) != 0:
print("last position:")
print("{0} {1} {2}".format(positions[-1].account_id,
positions[-1].stock_code, positions[-1].volume))
# 根据股票代码查询对应持仓
print("query position:")
position = xt_trader.query_stock_position(acc, stock_code)
if position:
print("position:")
print("{0} {1} {2}".format(position.account_id, position.stock_code,
position.volume))
# 阻塞线程，接收交易推送
xt_trader.run_forever()
开通xtquant的方式可以咨询。
目前开户费率低，门槛低，提供技术支持与交流。
需要的朋友，可以扫码咨询：
备注开户
python sqlite3 多线程 批量写入 【代码】
python • 李魔佛 发表了文章 • 0 个评论 • 3320 次浏览 • 2022-07-09 18:55
1. 随机生成一个数组数据
2. 在多线程里面批量插入数据
几个关注点：
sqlite3.connect(_type, check_same_thread=False) 要设置为False
批量写的时候，记得要加锁
import datetime
import random
import sqlite3
import threading
import logging as log
import time
lock = threading.Lock()
class SQLiteDBCls:
def __init__(self, cache=True):
_type = ":memory:"
self.db = sqlite3.connect(_type, check_same_thread=False)
self.table_name = 'tick_data'
def create_index(self):
cmd = 'CREATE INDEX code_ix ON {} (current)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def create_table(self):
# cursor = self.db.cursor()
cmd = 'create table if not exists {} (id INTEGER PRIMARY KEY AUTOINCREMENT,code text,open double,current time)'.format(
self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def add(self, code, price, t):
cmd = 'insert into {} (code,open,current) values (?,?,?);'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd, (code, price, t))
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def batch_add(self, data):
# 批量加入
print('===========',threading.current_thread().getName())
# log.info(threading.current_thread().getName())
cmd = 'insert into {} (code,open,current) values (?,?,?)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.executemany(cmd, data)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def result(self):
cmd = 'select count(*) from `{}`'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
return cursor.fetchone()
def data_gen():
minute = 6000
code = ['123011.SS','110010.SS','112111.SS']
for i in range(minute):
current = (datetime.datetime.now()+datetime.timedelta(minutes=i)).strftime('%H:%M:%D')
data_list =
for c in code:
price = 5+random.random()+120
data = (c,price,current)
data_list.append(data)
yield data_list
# time.sleep(0.5)
app = SQLiteDBCls(cache=True)
app.create_table()
app.create_index()
def data_validation():
print(app.result())
app.sync_up()
def multithread_mode():
total_count = 0
thread_list =
for d in data_gen():
print(d)
total_count+=len(d)
# app.batch_add(d)
t=threading.Thread(target=app.batch_add,args=(d,))
thread_list.append(t)
for t in thread_list:
t.start()
for t in thread_list:
t.join()
print(total_count)
if __name__=='__main__':
multithread_mode()
data_validation()
假如不加锁会出错：
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 910, in run
self._target(*self._args, **self._kwargs)
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 77, in batch_add
self.db.commit()
Exception in thread Thread-3824:
Exception in thread Thread-3826:
Traceback (most recent call last):
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 973, in _bootstrap_inner
sqlite3.OperationalError: cannot commit - no transaction is activeTraceback (most recent call last):
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 72, in batch_add
cursor.executemany(cmd, data)
sqlite3.InterfaceError: Error binding parameter 0 - probably unsupported type.
查看全部
2. 在多线程里面批量插入数据
几个关注点：
sqlite3.connect(_type, check_same_thread=False) 要设置为False
批量写的时候，记得要加锁
import datetime
import random
import sqlite3
import threading
import logging as log
import time
lock = threading.Lock()
class SQLiteDBCls:
def __init__(self, cache=True):
_type = ":memory:"
self.db = sqlite3.connect(_type, check_same_thread=False)
self.table_name = 'tick_data'
def create_index(self):
cmd = 'CREATE INDEX code_ix ON {} (current)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def create_table(self):
# cursor = self.db.cursor()
cmd = 'create table if not exists {} (id INTEGER PRIMARY KEY AUTOINCREMENT,code text,open double,current time)'.format(
self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def add(self, code, price, t):
cmd = 'insert into {} (code,open,current) values (?,?,?);'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd, (code, price, t))
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def batch_add(self, data):
# 批量加入
print('===========',threading.current_thread().getName())
# log.info(threading.current_thread().getName())
cmd = 'insert into {} (code,open,current) values (?,?,?)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.executemany(cmd, data)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def result(self):
cmd = 'select count(*) from `{}`'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
return cursor.fetchone()
def data_gen():
minute = 6000
code = ['123011.SS','110010.SS','112111.SS']
for i in range(minute):
current = (datetime.datetime.now()+datetime.timedelta(minutes=i)).strftime('%H:%M:%D')
data_list =
for c in code:
price = 5+random.random()+120
data = (c,price,current)
data_list.append(data)
yield data_list
# time.sleep(0.5)
app = SQLiteDBCls(cache=True)
app.create_table()
app.create_index()
def data_validation():
print(app.result())
app.sync_up()
def multithread_mode():
total_count = 0
thread_list =
for d in data_gen():
print(d)
total_count+=len(d)
# app.batch_add(d)
t=threading.Thread(target=app.batch_add,args=(d,))
thread_list.append(t)
for t in thread_list:
t.start()
for t in thread_list:
t.join()
print(total_count)
if __name__=='__main__':
multithread_mode()
data_validation()
假如不加锁会出错：
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 910, in run
self._target(*self._args, **self._kwargs)
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 77, in batch_add
self.db.commit()
Exception in thread Thread-3824:
Exception in thread Thread-3826:
Traceback (most recent call last):
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 973, in _bootstrap_inner
sqlite3.OperationalError: cannot commit - no transaction is activeTraceback (most recent call last):
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 72, in batch_add
cursor.executemany(cmd, data)
sqlite3.InterfaceError: Error binding parameter 0 - probably unsupported type.
查看全部
1. 随机生成一个数组数据
2. 在多线程里面批量插入数据
几个关注点：
sqlite3.connect(_type, check_same_thread=False) 要设置为False
批量写的时候，记得要加锁
假如不加锁会出错：
2. 在多线程里面批量插入数据
几个关注点：
sqlite3.connect(_type, check_same_thread=False) 要设置为False
批量写的时候，记得要加锁
import datetime
import random
import sqlite3
import threading
import logging as log
import time
lock = threading.Lock()
class SQLiteDBCls:
def __init__(self, cache=True):
_type = ":memory:"
self.db = sqlite3.connect(_type, check_same_thread=False)
self.table_name = 'tick_data'
def create_index(self):
cmd = 'CREATE INDEX code_ix ON {} (current)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def create_table(self):
# cursor = self.db.cursor()
cmd = 'create table if not exists {} (id INTEGER PRIMARY KEY AUTOINCREMENT,code text,open double,current time)'.format(
self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def add(self, code, price, t):
cmd = 'insert into {} (code,open,current) values (?,?,?);'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd, (code, price, t))
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def batch_add(self, data):
# 批量加入
print('===========',threading.current_thread().getName())
# log.info(threading.current_thread().getName())
cmd = 'insert into {} (code,open,current) values (?,?,?)'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.executemany(cmd, data)
except Exception as e:
log.info(e)
self.db.rollback()
else:
self.db.commit()
def result(self):
cmd = 'select count(*) from `{}`'.format(self.table_name)
with lock:
try:
cursor = self.db.cursor()
cursor.execute(cmd)
except Exception as e:
log.info(e)
self.db.rollback()
else:
return cursor.fetchone()
def data_gen():
minute = 6000
code = ['123011.SS','110010.SS','112111.SS']
for i in range(minute):
current = (datetime.datetime.now()+datetime.timedelta(minutes=i)).strftime('%H:%M:%D')
data_list =
for c in code:
price = 5+random.random()+120
data = (c,price,current)
data_list.append(data)
yield data_list
# time.sleep(0.5)
app = SQLiteDBCls(cache=True)
app.create_table()
app.create_index()
def data_validation():
print(app.result())
app.sync_up()
def multithread_mode():
total_count = 0
thread_list =
for d in data_gen():
print(d)
total_count+=len(d)
# app.batch_add(d)
t=threading.Thread(target=app.batch_add,args=(d,))
thread_list.append(t)
for t in thread_list:
t.start()
for t in thread_list:
t.join()
print(total_count)
if __name__=='__main__':
multithread_mode()
data_validation()
假如不加锁会出错：
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 910, in run
self._target(*self._args, **self._kwargs)
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 77, in batch_add
self.db.commit()
Exception in thread Thread-3824:
Exception in thread Thread-3826:
Traceback (most recent call last):
File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 973, in _bootstrap_inner
sqlite3.OperationalError: cannot commit - no transaction is activeTraceback (most recent call last):
File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 72, in batch_add
cursor.executemany(cmd, data)
sqlite3.InterfaceError: Error binding parameter 0 - probably unsupported type.
