Enhance trading workflow with new order flow management
- Added dingdanliu_nb_mflow for improved order processing - Updated related scripts and configurations to support new functionality
This commit is contained in:
184
2.数据下载与处理/update_data.py
Normal file
184
2.数据下载与处理/update_data.py
Normal file
@@ -0,0 +1,184 @@
|
||||
from multiprocessing import Process
|
||||
from datetime import datetime
|
||||
|
||||
from vnpy.trader.database import BarOverview
|
||||
from vnpy.trader.datafeed import get_datafeed
|
||||
from vnpy.trader.object import ContractData, BarData, HistoryRequest
|
||||
from vnpy.trader.constant import Exchange, Product, OptionType, Interval
|
||||
from vnpy.trader.setting import SETTINGS
|
||||
|
||||
from elite_database import EliteDatabase
|
||||
|
||||
|
||||
# 配置迅投研数据服务
|
||||
SETTINGS["datafeed.name"] = "xt"
|
||||
SETTINGS["datafeed.username"] = "token"
|
||||
SETTINGS["datafeed.password"] = ""
|
||||
|
||||
|
||||
# 交易所映射关系
|
||||
EXCHANGE_XT2VT = {
|
||||
"SH": Exchange.SSE,
|
||||
"SZ": Exchange.SZSE,
|
||||
"BJ": Exchange.BSE,
|
||||
"SF": Exchange.SHFE,
|
||||
"IF": Exchange.CFFEX,
|
||||
"INE": Exchange.INE,
|
||||
"DF": Exchange.DCE,
|
||||
"ZF": Exchange.CZCE,
|
||||
"GF": Exchange.GFEX
|
||||
}
|
||||
|
||||
|
||||
def update_history_data() -> None:
|
||||
"""更新历史合约信息"""
|
||||
# 在子进程中加载xtquant
|
||||
from xtquant.xtdata import download_history_data
|
||||
|
||||
# 初始化数据服务
|
||||
datafeed = get_datafeed()
|
||||
datafeed.init()
|
||||
|
||||
# 下载历史合约信息
|
||||
download_history_data("", "historycontract")
|
||||
|
||||
print("xtquant历史合约信息下载完成")
|
||||
|
||||
|
||||
def update_contract_data(sector_name: str) -> None:
|
||||
"""更新合约数据"""
|
||||
# 在子进程中加载xtquant
|
||||
from xtquant.xtdata import (
|
||||
get_stock_list_in_sector,
|
||||
get_instrument_detail
|
||||
)
|
||||
|
||||
# 初始化数据服务
|
||||
datafeed = get_datafeed()
|
||||
datafeed.init()
|
||||
|
||||
# 查询中金所历史合约代码
|
||||
vt_symbols: list[str] = get_stock_list_in_sector(sector_name)
|
||||
|
||||
# 遍历列表查询合约信息
|
||||
contracts: list[ContractData] = []
|
||||
|
||||
for xt_symbol in vt_symbols:
|
||||
# 拆分XT代码
|
||||
symbol, xt_exchange = xt_symbol.split(".")
|
||||
|
||||
# 筛选期权合约合约
|
||||
if "-" in symbol:
|
||||
data: dict = get_instrument_detail(xt_symbol, True)
|
||||
|
||||
type_str = data["InstrumentID"].split("-")[1]
|
||||
if type_str == "C":
|
||||
option_type = OptionType.CALL
|
||||
elif type_str == "P":
|
||||
option_type = OptionType.PUT
|
||||
|
||||
option_underlying: str = data["InstrumentID"].split("-")[0]
|
||||
|
||||
contract: ContractData = ContractData(
|
||||
symbol=data["InstrumentID"],
|
||||
exchange=EXCHANGE_XT2VT[xt_exchange.replace("O", "")],
|
||||
name=data["InstrumentName"],
|
||||
product=Product.OPTION,
|
||||
size=data["VolumeMultiple"],
|
||||
pricetick=data["PriceTick"],
|
||||
min_volume=data["MinLimitOrderVolume"],
|
||||
option_strike=data["ExtendInfo"]["OptExercisePrice"],
|
||||
option_listed=datetime.strptime(data["OpenDate"], "%Y%m%d"),
|
||||
option_expiry=datetime.strptime(data["ExpireDate"], "%Y%m%d"),
|
||||
option_underlying=option_underlying,
|
||||
option_portfolio=data["ProductID"],
|
||||
option_index=str(data["ExtendInfo"]["OptExercisePrice"]),
|
||||
option_type=option_type,
|
||||
gateway_name="XT"
|
||||
)
|
||||
contracts.append(contract)
|
||||
|
||||
# 保存合约信息到数据库
|
||||
database: EliteDatabase = EliteDatabase()
|
||||
database.save_contract_data(contracts)
|
||||
|
||||
print("合约信息更新成功", len(contracts))
|
||||
|
||||
|
||||
def update_bar_data() -> None:
|
||||
"""更新K线数据"""
|
||||
# 初始化数据服务
|
||||
datafeed = get_datafeed()
|
||||
datafeed.init()
|
||||
|
||||
# 获取当前时间戳
|
||||
now: datetime = datetime.now()
|
||||
|
||||
# 获取合约信息
|
||||
database: EliteDatabase = EliteDatabase()
|
||||
contracts: list[ContractData] = database.load_contract_data()
|
||||
|
||||
# 获取数据汇总
|
||||
data: list[BarOverview] = database.get_bar_overview()
|
||||
|
||||
overviews: dict[str, BarOverview] = {}
|
||||
for o in data:
|
||||
# 只保留分钟线数据
|
||||
if o.interval != Interval.MINUTE:
|
||||
continue
|
||||
|
||||
vt_symbol: str = f"{o.symbol}.{o.exchange.value}"
|
||||
overviews[vt_symbol] = o
|
||||
|
||||
# 遍历所有合约信息
|
||||
for contract in contracts:
|
||||
# 如果没有到期时间,则跳过
|
||||
if not contract.option_expiry:
|
||||
continue
|
||||
|
||||
# 查询数据汇总
|
||||
overview: BarOverview = overviews.get(contract.vt_symbol, None)
|
||||
|
||||
# 如果已经到期,则跳过
|
||||
if overview and contract.option_expiry < now:
|
||||
continue
|
||||
|
||||
# 初始化查询开始的时间
|
||||
start: datetime = datetime(2018, 1, 1)
|
||||
|
||||
# 实现增量查询
|
||||
if overview:
|
||||
start = overview.end
|
||||
|
||||
# 执行数据查询和更新入库
|
||||
req: HistoryRequest = HistoryRequest(
|
||||
symbol=contract.symbol,
|
||||
exchange=contract.exchange,
|
||||
start=start,
|
||||
end=datetime.now(),
|
||||
interval=Interval.MINUTE
|
||||
)
|
||||
|
||||
bars: list[BarData] = datafeed.query_bar_history(req)
|
||||
|
||||
if bars:
|
||||
database.save_bar_data(bars)
|
||||
|
||||
start_dt: datetime = bars[0].datetime
|
||||
end_dt: datetime = bars[-1].datetime
|
||||
msg: str = f"{contract.vt_symbol}数据更新成功,{start_dt} - {end_dt}"
|
||||
print(msg)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 使用子进程更新历史合约信息
|
||||
process: Process = Process(target=update_history_data)
|
||||
process.start()
|
||||
process.join() # 等待子进程执行完成
|
||||
|
||||
# 更新合约信息
|
||||
update_contract_data("中金所")
|
||||
update_contract_data("过期中金所")
|
||||
|
||||
# 更新历史数据
|
||||
# update_bar_data()
|
||||
Reference in New Issue
Block a user