679 lines
29 KiB
Python
679 lines
29 KiB
Python
# coding:utf-8
|
||
import time
|
||
import datetime
|
||
import threading
|
||
import logging
|
||
from xtquant import xtdata
|
||
from xtquant.xttrader import XtQuantTrader
|
||
from xtquant.xttype import StockAccount
|
||
from xtquant import xtconstant
|
||
from database_manager import DatabaseManager
|
||
from redis_state_manager import RedisStateManager
|
||
|
||
def is_call_auction_time(now_time=None):
|
||
"""
|
||
判断当前时间是否处于开盘集合竞价时段
|
||
"""
|
||
if not now_time:
|
||
now_time = datetime.datetime.now().time()
|
||
|
||
# 开盘集合竞价时间段
|
||
call_auction_start = datetime.time(9, 15)
|
||
call_auction_end = datetime.time(9, 25)
|
||
|
||
return call_auction_start <= now_time <= call_auction_end
|
||
|
||
def is_continuous_trading_time(now_time=None):
|
||
"""
|
||
判断当前时间是否处于连续交易时段
|
||
"""
|
||
if not now_time:
|
||
now_time = datetime.datetime.now().time()
|
||
|
||
# 上午连续交易时段
|
||
morning_start = datetime.time(9, 30)
|
||
morning_end = datetime.time(11, 30)
|
||
|
||
# 下午连续交易时段
|
||
afternoon_start = datetime.time(13, 0)
|
||
afternoon_end = datetime.time(14, 57)
|
||
|
||
is_morning = morning_start <= now_time <= morning_end
|
||
is_afternoon = afternoon_start <= now_time <= afternoon_end
|
||
|
||
return is_morning or is_afternoon
|
||
|
||
# 定义一个类 创建类的实例 作为状态的容器
|
||
class _a():
|
||
pass
|
||
|
||
A = _a()
|
||
A.hsa = xtdata.get_stock_list_in_sector('沪深A股')
|
||
A.order_timeout = 300 # 订单超时时间（秒），默认5分钟
|
||
|
||
# 数据库管理器实例
|
||
db_manager = None
|
||
redis_manager = None
|
||
|
||
def initialize_database_manager(logger=None):
|
||
"""初始化数据库管理器"""
|
||
global db_manager
|
||
if db_manager is None:
|
||
db_manager = DatabaseManager(logger)
|
||
return db_manager
|
||
|
||
def initialize_redis_manager(logger=None):
|
||
"""初始化Redis状态管理器"""
|
||
global redis_manager
|
||
if redis_manager is None:
|
||
redis_manager = RedisStateManager(logger)
|
||
return redis_manager
|
||
|
||
def load_trading_plans_from_database(strategy_id=1, logger=None):
|
||
"""已弃用：计划改为直接从Redis读取，保留此函数用于日志兼容"""
|
||
if logger:
|
||
logger.info("交易计划现改为从Redis读取，数据库仅做审计。")
|
||
return {}, {}
|
||
|
||
def load_initial_positions(xt_trader, acc, logger):
|
||
"""启动时一次性加载初始持仓状态（券商→Redis）并返回Redis视图"""
|
||
try:
|
||
# 先从QMT获取持仓
|
||
positions = xt_trader.query_stock_positions(acc)
|
||
qmt_positions = {pos.stock_code: pos.m_nVolume for pos in positions if pos.m_nVolume > 0}
|
||
|
||
# 从数据库获取持仓
|
||
db_manager = initialize_database_manager(logger)
|
||
db_positions = db_manager.get_current_positions()
|
||
|
||
# 将券商持仓写入Redis为权威值；DB中但券商为0的暂不写入，避免脏数据覆盖
|
||
r = initialize_redis_manager(logger)
|
||
# 清理Redis中不存在于券商的持仓（设为0即删除）
|
||
for code in list(r.get_all_positions().keys()):
|
||
if code not in qmt_positions:
|
||
r.set_position(code, 0)
|
||
for code, qty in qmt_positions.items():
|
||
r.set_position(code, qty)
|
||
|
||
logger.info(f"初始持仓加载完成，已同步至Redis")
|
||
logger.info(f" QMT持仓: {qmt_positions}")
|
||
logger.info(f" 数据库持仓: {db_positions}")
|
||
return r.get_all_positions()
|
||
except Exception as e:
|
||
logger.error(f"加载初始持仓失败: {str(e)}")
|
||
A.positions = {}
|
||
return {}
|
||
|
||
def update_position_in_memory(stock_code, quantity_change, is_buy=True, price=None, logger=None):
|
||
"""以Redis为主更新持仓，同时更新数据库（不再维护内存镜像）"""
|
||
try:
|
||
# Redis持仓更新
|
||
r = initialize_redis_manager(logger)
|
||
delta = quantity_change if is_buy else -quantity_change
|
||
new_qty = r.incr_position(stock_code, delta)
|
||
|
||
# 更新数据库持仓
|
||
if price is not None:
|
||
db_manager = initialize_database_manager(logger)
|
||
db_manager.update_position(stock_code, quantity_change, price, is_buy)
|
||
|
||
if logger:
|
||
logger.info(f"持仓状态已更新: {stock_code} -> {new_qty}股（Redis）")
|
||
except Exception as e:
|
||
if logger:
|
||
logger.error(f"更新持仓状态失败: {str(e)}")
|
||
|
||
def add_pending_order(stock_code, order_id, order_quantity, order_price, target_price, order_side='buy', logger=None):
|
||
"""添加在途订单：写Redis为主，同时记录数据库"""
|
||
try:
|
||
order_info = {
|
||
'order_id': order_id,
|
||
'order_quantity': order_quantity,
|
||
'order_price': order_price,
|
||
'target_price': target_price,
|
||
'order_time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||
'status': 'pending',
|
||
'side': order_side
|
||
}
|
||
# Redis
|
||
r = initialize_redis_manager(logger)
|
||
r.set_pending(stock_code, order_info)
|
||
|
||
# 记录到数据库
|
||
db_manager = initialize_database_manager(logger)
|
||
order_data = {
|
||
'order_id': order_id,
|
||
'strategy_id': 1,
|
||
'stock_code': stock_code,
|
||
'order_side': order_side,
|
||
'order_quantity': order_quantity,
|
||
'order_price': order_price,
|
||
'target_price': target_price,
|
||
'order_status': 'pending',
|
||
'order_time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||
'order_remark': f"{order_side.upper()}_{stock_code}_{int(time.time())}"
|
||
}
|
||
db_manager.insert_order(order_data)
|
||
|
||
if logger:
|
||
logger.info(f"添加在途订单: {stock_code} -> {order_id}")
|
||
except Exception as e:
|
||
if logger:
|
||
logger.error(f"添加在途订单失败: {str(e)}")
|
||
|
||
def remove_pending_order(stock_code, logger=None):
|
||
"""移除在途订单，同时更新数据库状态"""
|
||
try:
|
||
r = initialize_redis_manager(logger)
|
||
order_info = r.get_pending(stock_code)
|
||
|
||
if order_info is None:
|
||
if logger:
|
||
logger.warning(f"未找到在途订单: {stock_code}")
|
||
return
|
||
|
||
r.del_pending(stock_code)
|
||
|
||
# 更新数据库订单状态
|
||
print("====================================")
|
||
db_manager = initialize_database_manager(logger)
|
||
db_manager.update_order_status(order_info['order_id'], 'completed')
|
||
|
||
if logger:
|
||
logger.info(f"移除在途订单: {stock_code} -> {order_info['order_id']}")
|
||
except Exception as e:
|
||
if logger:
|
||
logger.error(f"移除在途订单失败: {str(e)}")
|
||
|
||
def is_stock_pending_order(stock_code):
|
||
"""检查股票是否有在途订单（以Redis为准）"""
|
||
r = initialize_redis_manager()
|
||
return r.get_pending(stock_code) is not None
|
||
|
||
def check_order_timeout(xt_trader, acc, logger):
|
||
"""已停用：本策略不再执行超时自动撤单，仅保留占位"""
|
||
logger.info("check_order_timeout 已停用：挂单后不再按超时撤单")
|
||
return
|
||
|
||
def start_order_timeout_monitor(xt_trader, acc, logger):
|
||
"""已停用：不再启动订单超时监控线程"""
|
||
logger.info("订单超时监控已停用：不启动后台监控线程")
|
||
return
|
||
|
||
def create_buy_strategy_callback(xt_trader, acc, buy_amount, logger):
|
||
"""创建买入策略回调函数"""
|
||
def f(data):
|
||
"""
|
||
实时行情回调函数
|
||
检查当前价格是否低于目标价格，如果满足条件则买入
|
||
"""
|
||
logger.info(f"收到买入行情数据: {datetime.datetime.now()}")
|
||
|
||
# 获取当前时间
|
||
current_time = datetime.datetime.now().time()
|
||
|
||
# 判断当前交易时段
|
||
is_call_auction = is_call_auction_time(current_time)
|
||
is_continuous = is_continuous_trading_time(current_time)
|
||
|
||
for stock_code, current_data in data.items():
|
||
if stock_code not in A.hsa:
|
||
continue
|
||
|
||
# 从Redis读取买入计划
|
||
r = initialize_redis_manager(logger)
|
||
plan_info = r.get_buy_plan(stock_code)
|
||
if not plan_info or plan_info.get('is_active', 1) != 1:
|
||
continue
|
||
|
||
try:
|
||
# 获取当前价格和目标价格
|
||
current_price = current_data[0]['close']
|
||
target_price = plan_info['target_price']
|
||
buy_amount = plan_info['buy_amount']
|
||
|
||
logger.info(f"[买入监控] {stock_code} 当前价格: {current_price:.2f}, 目标价格: {target_price:.2f}")
|
||
|
||
# 判断是否满足买入条件：当前价格低于目标价格
|
||
if current_price < target_price:
|
||
|
||
# 检查持仓（Redis）
|
||
current_qty = r.get_position(stock_code)
|
||
if current_qty > 0:
|
||
logger.info(f"{stock_code} 已有持仓 {current_qty}股，跳过买入")
|
||
continue
|
||
|
||
# 检查是否有在途订单（Redis）
|
||
if is_stock_pending_order(stock_code):
|
||
logger.info(f"{stock_code} 有在途订单，跳过买入")
|
||
continue
|
||
|
||
# 集合竞价时段：只观察，不下单
|
||
if is_call_auction:
|
||
logger.info(f"[集合竞价观察] {stock_code} 当前价格 {current_price:.2f} < 目标价格 {target_price:.2f}，等待连续交易时段下单")
|
||
continue
|
||
|
||
# 连续交易时段：执行下单
|
||
if is_continuous:
|
||
logger.info(f"[连续交易下单] 满足买入条件！{stock_code} 当前价格 {current_price:.2f} < 目标价格 {target_price:.2f}")
|
||
|
||
# 计算买入数量（从数据库获取金额）
|
||
buy_volume = int(buy_amount / current_price / 100) * 100 # 取整为100的整数倍
|
||
|
||
if buy_volume > 0:
|
||
logger.info(f"准备买入 {stock_code} {buy_volume}股，金额约 {buy_volume * current_price:.2f}元")
|
||
|
||
# 生成订单ID
|
||
order_id = f"BUY_{stock_code}_{int(time.time())}"
|
||
|
||
# 执行买入订单
|
||
async_seq = xt_trader.order_stock_async(
|
||
acc,
|
||
stock_code,
|
||
xtconstant.STOCK_BUY,
|
||
buy_volume,
|
||
xtconstant.FIX_PRICE,
|
||
current_price,
|
||
'auto_buy_strategy',
|
||
order_id
|
||
)
|
||
|
||
# 添加到在途订单（同时记录到数据库）
|
||
add_pending_order(stock_code, order_id, buy_volume, current_price, target_price, 'buy', logger)
|
||
|
||
# 单次更新账户资金（不阻塞回调）
|
||
update_account_funds_once(xt_trader, acc, logger)
|
||
|
||
logger.info(f"已提交买入订单: {stock_code} {buy_volume}股，价格{target_price}，订单ID: {order_id}")
|
||
else:
|
||
logger.warning(f"买入数量计算为0，跳过 {stock_code}")
|
||
else:
|
||
# 非交易时段
|
||
logger.info(f"[非交易时段] {stock_code} 当前价格 {current_price:.2f} < 目标价格 {target_price:.2f}，等待交易时段")
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理 {stock_code} 买入行情数据时出错: {str(e)}")
|
||
continue
|
||
|
||
return f
|
||
|
||
def create_sell_strategy_callback(xt_trader, acc, logger):
|
||
"""创建卖出策略回调函数"""
|
||
def f(data):
|
||
"""
|
||
实时行情回调函数
|
||
检查当前价格是否高于目标价格，如果满足条件且持有该股票则卖出
|
||
"""
|
||
# logger.info(f"收到卖出行情数据: {datetime.datetime.now()}")
|
||
|
||
# 获取当前时间
|
||
current_time = datetime.datetime.now().time()
|
||
|
||
# 判断当前交易时段
|
||
is_call_auction = is_call_auction_time(current_time)
|
||
is_continuous = is_continuous_trading_time(current_time)
|
||
|
||
for stock_code, current_data in data.items():
|
||
if stock_code not in A.hsa:
|
||
continue
|
||
|
||
# 从Redis读取卖出计划
|
||
r = initialize_redis_manager(logger)
|
||
plan_info = r.get_sell_plan(stock_code)
|
||
if not plan_info or plan_info.get('is_active', 1) != 1:
|
||
continue
|
||
|
||
try:
|
||
# 获取当前价格和目标价格
|
||
current_price = current_data[0]['close']
|
||
target_price = plan_info['target_price']
|
||
|
||
# logger.info(f"[卖出监控] {stock_code} 当前价格: {current_price:.2f}, 目标价格: {target_price:.2f}")
|
||
|
||
# 判断是否满足卖出条件：当前价格高于目标价格
|
||
if current_price > target_price:
|
||
|
||
# 检查持仓（Redis）
|
||
r = initialize_redis_manager(logger)
|
||
current_position = r.get_position(stock_code)
|
||
if current_position <= 0:
|
||
logger.info(f"{stock_code} 无持仓，跳过卖出")
|
||
continue
|
||
|
||
# 检查是否有在途订单
|
||
if is_stock_pending_order(stock_code):
|
||
# logger.info(f"{stock_code} 有在途订单，跳过卖出")
|
||
continue
|
||
|
||
# 集合竞价时段：只观察，不下单
|
||
if is_call_auction:
|
||
logger.info(f"[集合竞价观察] {stock_code} 当前价格 {current_price:.2f} > 目标价格 {target_price:.2f}，等待连续交易时段下单")
|
||
continue
|
||
|
||
# 连续交易时段：执行下单
|
||
if is_continuous:
|
||
logger.info(f"[连续交易下单] 满足卖出条件！{stock_code} 当前价格 {current_price:.2f} > 目标价格 {target_price:.2f}")
|
||
|
||
# 卖出数量：全部持仓
|
||
sell_volume = current_position
|
||
|
||
if sell_volume > 0:
|
||
logger.info(f"准备卖出 {stock_code} {sell_volume}股，金额约 {sell_volume * current_price:.2f}元")
|
||
|
||
# 生成订单ID
|
||
order_id = f"SELL_{stock_code}_{int(time.time())}"
|
||
# 执行卖出订单
|
||
async_seq = xt_trader.order_stock_async(
|
||
acc,
|
||
stock_code,
|
||
xtconstant.STOCK_SELL,
|
||
sell_volume,
|
||
xtconstant.LATEST_PRICE,
|
||
target_price,
|
||
'auto_sell_strategy',
|
||
order_id
|
||
)
|
||
|
||
# 添加到在途订单（同时记录到数据库）
|
||
add_pending_order(stock_code, order_id, sell_volume, current_price, target_price, 'sell', logger)
|
||
|
||
logger.info(f"已提交卖出订单: {stock_code} {sell_volume}股，价格{target_price}，订单ID: {order_id}")
|
||
# 单次更新账户资金（不阻塞回调）
|
||
update_account_funds_once(xt_trader, acc, logger)
|
||
else:
|
||
logger.warning(f"卖出数量为0，跳过 {stock_code}")
|
||
else:
|
||
# 非交易时段
|
||
logger.info(f"[非交易时段] {stock_code} 当前价格 {current_price:.2f} > 目标价格 {target_price:.2f}，等待交易时段")
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理 {stock_code} 卖出行情数据时出错: {str(e)}")
|
||
continue
|
||
|
||
return f
|
||
|
||
def _update_account_funds_sync(xt_trader, acc, logger):
|
||
"""同步更新账户资金信息（内部函数，在后台线程中执行）"""
|
||
try:
|
||
# 获取账户资金信息
|
||
account_info = xt_trader.query_stock_asset(acc)
|
||
|
||
if account_info:
|
||
# 计算相关数据
|
||
available_cash = getattr(account_info, 'm_dCash', 0.0)
|
||
frozen_cash = getattr(account_info, 'm_dFrozenCash', 0.0)
|
||
market_value = getattr(account_info, 'm_dMarketValue', 0.0)
|
||
total_asset = available_cash + frozen_cash + market_value
|
||
|
||
# 尝试获取盈亏信息，如果属性不存在则设为0
|
||
try:
|
||
profit_loss = getattr(account_info, 'm_dProfitLoss', 0.0)
|
||
except AttributeError:
|
||
# 如果m_dProfitLoss不存在，尝试其他可能的属性名
|
||
profit_loss = getattr(account_info, 'm_dProfit', 0.0)
|
||
|
||
# 计算盈亏比例
|
||
if total_asset > 0:
|
||
profit_loss_ratio = (profit_loss / total_asset) * 100
|
||
else:
|
||
profit_loss_ratio = 0.0
|
||
|
||
# 更新到数据库
|
||
db_manager = initialize_database_manager(logger)
|
||
success = db_manager.update_account_funds(
|
||
account_id=acc.account_id,
|
||
account_type=acc.account_type, # 修复：去掉引号
|
||
total_asset=total_asset,
|
||
available_cash=available_cash,
|
||
frozen_cash=frozen_cash,
|
||
market_value=market_value,
|
||
profit_loss=profit_loss,
|
||
profit_loss_ratio=profit_loss_ratio
|
||
)
|
||
|
||
if success:
|
||
logger.info(f"账户资金更新成功 - 总资产: {total_asset:.2f}, 可用资金: {available_cash:.2f}, 市值: {market_value:.2f}")
|
||
else:
|
||
logger.warning("账户资金更新失败")
|
||
else:
|
||
logger.warning("无法获取账户资金信息")
|
||
|
||
except Exception as e:
|
||
logger.error(f"更新账户资金信息时出错: {str(e)}")
|
||
|
||
def update_account_funds_once(xt_trader, acc, logger):
|
||
"""单次更新账户资金信息到数据库（异步执行，不阻塞回调函数）"""
|
||
# 在后台线程中执行，避免阻塞行情回调
|
||
thread = threading.Thread(
|
||
target=_update_account_funds_sync,
|
||
args=(xt_trader, acc, logger),
|
||
daemon=True
|
||
)
|
||
thread.start()
|
||
|
||
def update_account_funds_periodically(xt_trader, acc, logger):
|
||
"""每小时更新账户资金信息到数据库（用于后台线程，包含循环）"""
|
||
while True:
|
||
try:
|
||
# 获取账户资金信息
|
||
account_info = xt_trader.query_stock_asset(acc)
|
||
|
||
if account_info:
|
||
# 计算相关数据
|
||
available_cash = getattr(account_info, 'm_dCash', 0.0)
|
||
frozen_cash = getattr(account_info, 'm_dFrozenCash', 0.0)
|
||
market_value = getattr(account_info, 'm_dMarketValue', 0.0)
|
||
total_asset = available_cash + frozen_cash + market_value
|
||
|
||
# 尝试获取盈亏信息，如果属性不存在则设为0
|
||
try:
|
||
profit_loss = getattr(account_info, 'm_dProfitLoss', 0.0)
|
||
except AttributeError:
|
||
# 如果m_dProfitLoss不存在，尝试其他可能的属性名
|
||
profit_loss = getattr(account_info, 'm_dProfit', 0.0)
|
||
|
||
# 计算盈亏比例
|
||
if total_asset > 0:
|
||
profit_loss_ratio = (profit_loss / total_asset) * 100
|
||
else:
|
||
profit_loss_ratio = 0.0
|
||
|
||
# 更新到数据库
|
||
db_manager = initialize_database_manager(logger)
|
||
success = db_manager.update_account_funds(
|
||
account_id=acc.account_id,
|
||
account_type=acc.account_type, # 修复：去掉引号
|
||
total_asset=total_asset,
|
||
available_cash=available_cash,
|
||
frozen_cash=frozen_cash,
|
||
market_value=market_value,
|
||
profit_loss=profit_loss,
|
||
profit_loss_ratio=profit_loss_ratio
|
||
)
|
||
|
||
if success:
|
||
logger.info(f"账户资金更新成功 - 总资产: {total_asset:.2f}, 可用资金: {available_cash:.2f}, 市值: {market_value:.2f}")
|
||
else:
|
||
logger.warning("账户资金更新失败")
|
||
else:
|
||
logger.warning("无法获取账户资金信息")
|
||
|
||
except Exception as e:
|
||
logger.error(f"更新账户资金信息时出错: {str(e)}")
|
||
|
||
# 等待1小时（3600秒）
|
||
time.sleep(3600)
|
||
|
||
def start_account_funds_monitor(xt_trader, acc, logger):
|
||
"""启动账户资金监控线程（每小时更新一次）"""
|
||
try:
|
||
funds_thread = threading.Thread(
|
||
target=update_account_funds_periodically,
|
||
args=(xt_trader, acc, logger),
|
||
daemon=True
|
||
)
|
||
funds_thread.start()
|
||
logger.info("账户资金监控线程已启动（每小时更新一次）")
|
||
except Exception as e:
|
||
logger.error(f"启动账户资金监控失败: {str(e)}")
|
||
|
||
# ========== 每5分钟对账：查询券商持仓并反向更新订单/持仓 ==========
|
||
def reconcile_orders_and_positions(xt_trader, acc, logger):
|
||
"""一次性对账（基于当日成交对比委托）：
|
||
- 仅查询MySQL今日订单（submitted、filled[部分]、pending、ordered）
|
||
- 通过QMT接口获取当日成交明细/汇总，按股票累计成交量对比委托量：
|
||
* 当日成交量 >= 委托量 => 状态置为 completed
|
||
* 0 < 当日成交量 < 委托量 => 状态置为 filled（部分成交），记录 filled_quantity
|
||
* 当日成交量 == 0 => 不处理
|
||
- 本函数不更新持仓，持仓改由快照线程覆盖更新
|
||
"""
|
||
try:
|
||
dbm = initialize_database_manager(logger)
|
||
# 1) 读取今日未完成订单（四种状态：submitted、filled(部分成交)、pending、委托订单）
|
||
# 你提到“委托订单”第四种状态名，这里按 'ordered' 兜底；如你表里具体值不同，可替换。
|
||
candidate_statuses = ['submitted', 'filled', 'pending', 'ordered']
|
||
orders = dbm.get_todays_orders_by_statuses(candidate_statuses)
|
||
|
||
if not orders:
|
||
return
|
||
|
||
# 2) 通过QMT查询当日成交结果，并按股票聚合成交数量
|
||
filled_volume_by_code = {}
|
||
trades = xt_trader.query_stock_orders(acc, cancelable_only = False)
|
||
|
||
for tr in trades or []:
|
||
code = getattr(tr, 'stock_code', None) or getattr(tr, 'm_strInstrumentID', None)
|
||
vol = int(getattr(tr, 'traded_volume', 0) or getattr(tr, 'm_nVolume', 0) or 0)
|
||
if not code or vol <= 0:
|
||
continue
|
||
filled_volume_by_code[code] = filled_volume_by_code.get(code, 0) + vol
|
||
|
||
# 3) 遍历订单并根据“当日成交累计 vs 委托量”判断状态
|
||
for od in orders:
|
||
try:
|
||
stock_code = od['stock_code']
|
||
side = od['side']
|
||
order_id = od['order_id']
|
||
qty = int(od.get('order_quantity') or 0)
|
||
day_filled = int(filled_volume_by_code.get(stock_code, 0))
|
||
|
||
if day_filled <= 0:
|
||
continue
|
||
|
||
if day_filled >= qty and qty > 0:
|
||
# 全部成交 -> completed
|
||
dbm.update_order_status(order_id, 'completed', qty, datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
|
||
logger.info(f"[对账-简] 全部成交(>=委托): {stock_code} {side} {day_filled}/{qty}")
|
||
elif 0 < day_filled < qty:
|
||
# 部分成交 -> filled（部分成交）
|
||
dbm.update_order_status(order_id, 'filled', day_filled, datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
|
||
logger.info(f"[对账-简] 部分成交(<委托): {stock_code} {side} {day_filled}/{qty}")
|
||
except Exception as ie:
|
||
logger.warning(f"[对账-简] 处理订单失败 {od}: {ie}")
|
||
|
||
except Exception as e:
|
||
if logger:
|
||
logger.error(f"对账过程出错: {str(e)}")
|
||
|
||
def start_reconciliation_monitor(xt_trader, acc, logger, interval_seconds: int = 300):
|
||
"""启动对账后台线程（默认每5分钟一次）"""
|
||
try:
|
||
def worker():
|
||
while True:
|
||
try:
|
||
reconcile_orders_and_positions(xt_trader, acc, logger)
|
||
except Exception as ie:
|
||
logger.warning(f"对账线程异常: {ie}")
|
||
time.sleep(interval_seconds)
|
||
|
||
t = threading.Thread(target=worker, daemon=True)
|
||
t.start()
|
||
logger.info(f"对账线程已启动（每{interval_seconds}秒）")
|
||
except Exception as e:
|
||
logger.error(f"启动对账线程失败: {str(e)}")
|
||
|
||
# ========== 每5分钟持仓快照覆盖更新（仅DB） ==========
|
||
def refresh_positions_snapshot(xt_trader, acc, logger):
|
||
"""从QMT拉取当前持仓，覆盖更新到DB的 trading_position：
|
||
- 对存在的标的 upsert 所有持仓字段（数量、可用、冻结、成本价、市价、市值、盈亏等）
|
||
- 删除DB中多余的标的
|
||
本函数不操作Redis。
|
||
"""
|
||
try:
|
||
dbm = initialize_database_manager(logger)
|
||
positions = xt_trader.query_stock_positions(acc) or []
|
||
valid_codes = []
|
||
for pos in positions:
|
||
try:
|
||
code = pos.stock_code
|
||
# 获取持仓基本信息
|
||
total_qty = int(getattr(pos, 'volume', 0)) # 使用 'volume' 或 'm_nVolume'
|
||
available_qty = int(getattr(pos, 'can_use_volume', 0)) # 使用 'can_use_volume' 或 'm_nCanUseVolume'
|
||
|
||
# 直接使用API返回的冻结数量字段，这比自己计算更准确
|
||
frozen_qty = int(getattr(pos, 'frozen_volume', 0)) # 使用 'frozen_volume' 或 'm_nFrozenVolume'
|
||
|
||
# 如果总持仓为0，则没有处理的必要，直接跳过
|
||
if total_qty == 0:
|
||
continue
|
||
|
||
# 2. 获取价格和市值信息
|
||
cost_price = float(getattr(pos, 'avg_price', 0.0)) # 使用 'avg_price' 或 'm_dAvgPrice'
|
||
market_value = float(getattr(pos, 'market_value', 0.0)) # 使用 'market_value' 或 'm_dMarketValue'
|
||
|
||
# 3. 根据现有字段计算缺失的信息
|
||
|
||
# 计算市价 (Market Price)
|
||
# 正如您所说，通过 市值 / 总数量 来计算，并处理总数量为0的情况
|
||
market_price = market_value / total_qty if total_qty > 0 else 0.0
|
||
|
||
# 计算持仓成本 (Position Cost)
|
||
position_cost = cost_price * total_qty
|
||
|
||
# 计算持仓盈亏 (Profit/Loss)
|
||
# 盈亏 = 总市值 - 总成本
|
||
profit_loss = market_value - position_cost
|
||
|
||
# 计算盈亏比例 (Profit/Loss Ratio)
|
||
if position_cost > 0:
|
||
profit_loss_ratio = (profit_loss / position_cost) * 100
|
||
else:
|
||
profit_loss_ratio = 0.0 # 如果成本为0，则盈亏比例也为0
|
||
|
||
# if total_qty > 0:
|
||
dbm.upsert_position_snapshot(
|
||
code, total_qty, available_qty, frozen_qty,
|
||
cost_price, market_price, market_value,
|
||
profit_loss, profit_loss_ratio
|
||
)
|
||
valid_codes.append(code)
|
||
except Exception as e:
|
||
logger.warning(f"处理持仓记录失败 {getattr(pos, 'stock_code', 'unknown')}: {e}")
|
||
continue
|
||
# 清理非持有标的
|
||
dbm.delete_positions_except(valid_codes)
|
||
logger.info(f"[持仓快照] 已覆盖更新 {len(valid_codes)} 条持仓记录")
|
||
except Exception as e:
|
||
if logger:
|
||
logger.error(f"持仓快照刷新失败: {str(e)}")
|
||
|
||
def start_position_snapshot_monitor(xt_trader, acc, logger, interval_seconds: int = 300):
|
||
try:
|
||
def worker():
|
||
while True:
|
||
try:
|
||
refresh_positions_snapshot(xt_trader, acc, logger)
|
||
except Exception as ie:
|
||
logger.warning(f"持仓快照线程异常: {ie}")
|
||
time.sleep(interval_seconds)
|
||
t = threading.Thread(target=worker, daemon=True)
|
||
t.start()
|
||
logger.info(f"持仓快照线程已启动（每{interval_seconds}秒）")
|
||
except Exception as e:
|
||
logger.error(f"启动持仓快照线程失败: {str(e)}")
|
||
|
||
# 为了向后兼容，保留原来的函数名
|
||
def create_strategy_callback(xt_trader, acc, buy_amount, logger):
|
||
"""创建策略回调函数（兼容性函数，实际调用买入回调）"""
|
||
return create_buy_strategy_callback(xt_trader, acc, buy_amount, logger)
|