"""
|
交易方式模块（总览处理所有渠道的各种交易方法集合）
|
"""
|
import logging
|
import multiprocessing
|
import threading
|
|
from db import redis_manager_delegate as redis_manager
|
from db.redis_manager_delegate import RedisUtils
|
from log_module import async_log_util
|
from strategy import data_cache, account_management
|
import data_server
|
from log_module.log import logger_debug, logger_common, logger_trade
|
from strategy.trade_setting import TradeSetting
|
|
from trade import huaxin_trade_api, huaxin_trade_data_update, middle_api_protocol
|
from utils import huaxin_util, tool
|
|
# 引入日志模块
|
# 获取logger实例
|
logger = logger_common
|
|
|
# 下单买入函数（按金额，以限价买）【按金额买 基础版】
|
def buy_order_by_value(symbol, buy_order_value, sec_name, current_price):
|
if symbol[-6:] in TodayBuyCodeManager().get_buy_codes():
|
logger.info(f"{symbol}已经下过单")
|
return
|
|
# 自动买 开关监听方法
|
if not TradeSetting().get_auto_buy():
|
# 暂停自动买
|
logger.info(f"在交易方法函数处 关闭了 自动买")
|
return
|
# 限制交易标的的单价范围
|
if current_price < 3 or current_price > 30:
|
# 当前单价超出预设限制
|
logger.info(f"当前标的个股【{sec_name}】单价超出预设限制！预设值3 < current_price < 30,当前最新价{current_price}")
|
return
|
price = round(float(current_price), 2)
|
volume = (int(buy_order_value / price) // 100) * 100
|
if volume < 100:
|
volume = 100
|
# 调用笼子价计算工具计算下单价格
|
order_price = round(tool.get_buy_max_price(current_price), 2)
|
buy_order = huaxin_trade_api.order(1, symbol[-6:], volume, order_price, blocking=True)
|
logger.info(f"current_price===={current_price} order_price===={order_price}")
|
logger.info(f"buy_order===={buy_order}")
|
orderStatusMsg = buy_order['data'].get('orderStatusMsg', None)
|
orderRef = buy_order['data'].get('orderRef', None)
|
statusMsg = buy_order['data'].get('statusMsg', None)
|
logger.info(f"orderStatusMsg==={orderStatusMsg}")
|
logger.info(f"statusMsg==={statusMsg}")
|
# orderStatusMsg 不在buy_order的下单回调中，那么才认为下单成功
|
if statusMsg is not None and statusMsg == '':
|
TodayBuyCodeManager().place_order(symbol[-6:], orderRef)
|
logger.info(f"买票 下单成功：【{sec_name}】")
|
# 每一次成功下单后要更新一下 缓存 的持仓数据
|
account_management.position_management()
|
logger.info(f"更新的持仓数据data_cache.account_positions_dict=={data_cache.account_positions_dict}")
|
# 调用资金查询函数 查看资金变化
|
account_management.finance_management()
|
logger.info(f"更新的资金数据data_cache.account_finance_dict=={data_cache.account_finance_dict}")
|
if symbol[-6:] in data_cache.account_positions_dict:
|
logger.info(f"该股已经持仓==》{sec_name}")
|
pass
|
|
# 检测持仓信息中有无下单个股且有该个股的当前持仓，只有当前持仓数量不为0时，才认为交易成功
|
for i in data_cache.account_positions_dict:
|
# print(i)
|
if i['securityID'] == symbol[-6:]:
|
# print(i['currentPosition'])
|
if i['currentPosition'] == 0:
|
logger.info(f"【{i['securityName']}】交易失败~")
|
else:
|
# 买票后添加 持仓代码集合
|
data_cache.position_symbols_set.add(symbol[-6:])
|
logger.info(f"【{i['securityName']}】交易成功！")
|
|
|
# 下单买入函数（按可用资金的一定比例，在涨停价买）【按金额买 高级版】
|
def buy_order_by_part_value(part_of_value, symbol, available, today_limit_up_price, sec_name, index):
|
"""
|
:param symbol: 代码
|
:param available: 可用资金
|
:param part_of_value: 计划委买 账户余额的 比例
|
:param today_limit_up_price: 今日涨停价
|
:param sec_name: 公司名称
|
:param index: 持仓对象列表中的个股对应序列号
|
:return: 尝试返回的订单数据
|
"""
|
logger.info(f"当前账户可用资金available==={available}")
|
buy_order_value = round(available * part_of_value, 2)
|
logger.info(f"当前计划比例==={part_of_value}，当前委托金额==={buy_order_value}")
|
# 只有持仓数量大于委卖数量才进入买票流程
|
if available >= buy_order_value:
|
sell_order_by_volume(symbol, buy_order_value, sec_name, today_limit_up_price)
|
logger.info(f"【十分之{part_of_value * 10}可用资金】委买完毕")
|
data_cache.available = available - buy_order_value
|
logger.info(f"买票执行成功：【{sec_name}】")
|
logger.info(f"买票后剩余资金：{data_cache.account_finance_dict['usefulMoney']}")
|
else:
|
logger.info(f"【{sec_name}】，持仓：{available}小于计划委托：{part_of_value},委托失败!")
|
|
|
# 下单卖出函数（按持仓数量，在限价卖）【按量卖 基础版】
|
def sell_order_by_volume(symbol, volume, sec_name, current_price):
|
# 自动卖开关监听方法
|
if not TradeSetting().get_auto_sell():
|
# 暂停自动卖
|
logger.info(f"在交易方法函数处 关闭了 自动卖")
|
return
|
# price = round(float(price), 2)
|
# 调用笼子价计算工具计算下单价格
|
order_price = tool.get_buy_min_price(current_price)
|
sell_order = huaxin_trade_api.order(2, symbol[-6:], volume, order_price, blocking=True)
|
logger.info(f"current_price===={current_price} order_price===={order_price}")
|
logger.info(f"sell_order===={sell_order}")
|
orderStatusMsg = sell_order['data'].get('orderStatusMsg', None)
|
statusMsg = sell_order['data'].get('statusMsg', None)
|
logger.info(f"orderStatusMsg==={orderStatusMsg}")
|
logger.info(f"statusMsg==={statusMsg}")
|
# orderStatusMsg 不在buy_order的下单回调中，那么才认为下单成功
|
if statusMsg is not None and statusMsg == '':
|
logger.info(f"卖票 下单成功：【{sec_name}】")
|
# 每一次成功下单后要更新一下 缓存 的持仓数据
|
account_management.position_management()
|
logger.info(f"更新的持仓数据data_cache.account_positions_dict=={data_cache.account_positions_dict}")
|
# 调用资金查询函数 查看资金变化
|
account_management.finance_management()
|
logger.info(f"更新的资金数据data_cache.account_finance_dict=={data_cache.account_finance_dict}")
|
|
|
# 下单卖出函数（按持仓数量的一定比例，在跌停价卖）【按量卖 高级版】
|
def sell_order_by_part_volume(part_of_volume, symbol, position_volume_yesterday, current_price, sec_name,
|
index):
|
"""
|
:param symbol: 代码
|
:param position_volume_yesterday: 可用持仓数量
|
:param part_of_volume: 计划委卖持仓量的比例
|
:param current_price: 当前最新价
|
:param sec_name: 公司名称
|
:param index: 持仓对象列表中的个股对应序列号
|
:return: 尝试返回的订单数据
|
"""
|
|
logger.info(
|
f"当前个股持仓手数【当前函数被调用时传进来的同步数据data_cache中的持仓数据】==={position_volume_yesterday}")
|
# sell_order_volume = int(position_volume_yesterday * part_of_volume)
|
sell_order_volume = round(position_volume_yesterday * part_of_volume / 100) * 100
|
logger.info(f"当前计划比例==={part_of_volume}，当前委托量==={sell_order_volume}")
|
|
# 当委托量大于0
|
if sell_order_volume > 0:
|
# 只有持仓数量大于委卖数量才进入买票流程
|
if position_volume_yesterday >= sell_order_volume:
|
sell_order_by_volume(symbol, sell_order_volume, sec_name, current_price)
|
logger.info(f"【十分之 {round(part_of_volume * 10)} 仓】委卖完毕")
|
# 计算并更新剩余可用持仓数量
|
# data_cache.account_positions[index]['volume'] = position_volume_yesterday - sell_order_volume
|
if data_cache.account_positions_dict[index]['currentPosition'] <= 0:
|
logger.info(f"data_cache.account_positions == {data_cache.account_positions_dict}")
|
logger.info(
|
f"下单后，【{sec_name}】的剩余可用持仓数量==={data_cache.account_positions_dict[index]['currentPosition']}")
|
# 本票本次卖票，可用仓位为0或小于0，,移除【可用持仓代码】集合
|
'''
|
全局变量中的可用个股数量，由于只在【集合竞价】阶段用，如果移除会影响进入次数，暂不考虑使用
|
'''
|
# data_cache.available_symbols_set.remove(symbol)
|
logger.info(f"【{sec_name}】当日可用仓位数卖完了")
|
logger.info(f"卖后可用持仓票数:::{len(data_cache.available_symbols_set)}")
|
else:
|
logger.info(
|
f"【{sec_name}】，可用持仓：{position_volume_yesterday}小于计划委托：{sell_order_volume},无法委托，直接平仓!")
|
sell_order_by_volume(symbol, position_volume_yesterday, sec_name, current_price)
|
# 计算并更新剩余可用持仓数量
|
# data_cache.account_positions_dict[index]['currentPosition'] = position_volume_yesterday - position_volume_yesterday
|
else:
|
logger.info(f"委托量小于等于零，委托失败!")
|
logger.info(
|
f"【{sec_name}】，可用持仓：{position_volume_yesterday}，计划委托：{sell_order_volume}<=0 ？,无法委托，直接委卖100!")
|
sell_order_by_volume(symbol, 100, sec_name, current_price)
|
|
|
def run(queue_strategy_r_trade_w, queue_strategy_w_trade_r, queue_strategy_w_trade_for_query_r):
|
class MyTradeCallback(huaxin_trade_api.TradeCallback):
|
def on_order(self, order_info):
|
"""
|
订单状态改变回调
|
:param order_info: {'sinfo': 'b_603682_1736312765623', 'securityID': '603682', 'orderLocalID': '8100043081', 'direction': '0', 'orderSysID': '110018100043081', 'insertTime': '13:06:04', 'insertDate': '20250108', 'acceptTime': '13:05:46', 'cancelTime': '', 'limitPrice': 6.45, 'accountID': '00032047', 'orderRef': 130608, 'turnover': 6410.0, 'volume': 1000, 'volumeTraded': 1000, 'orderStatus': '4', 'orderSubmitStatus': '1', 'statusMsg': ''}
|
:return:
|
"""
|
print(f"收到订单回调：{order_info}")
|
async_log_util.info(logger_debug, f"收到订单回调：{order_info}")
|
if huaxin_util.is_deal(order_info['orderStatus']):
|
if order_info["direction"] == '0':
|
# 买入成交
|
TodayBuyCodeManager().add_deal_code(order_info["securityID"], order_info.get("orderRef"))
|
# 成交，需要更新持仓/委托/成交
|
huaxin_trade_data_update.add_position_list()
|
huaxin_trade_data_update.add_delegate_list("成交")
|
huaxin_trade_data_update.add_deal_list()
|
else:
|
huaxin_trade_data_update.add_money_list()
|
huaxin_trade_data_update.add_delegate_list("订单状态变化")
|
# 推送订单数据
|
threading.Thread(target=lambda: middle_api_protocol.push(
|
middle_api_protocol.load_push_msg({"type": "order", "data": order_info})), daemon=True).start()
|
|
huaxin_trade_api.run_trade(queue_strategy_r_trade_w, MyTradeCallback(), queue_strategy_w_trade_r,
|
queue_strategy_w_trade_for_query_r)
|
threading.Thread(target=data_server.run, daemon=True).start()
|
|
|
@tool.singleton
|
class TodayBuyCodeManager:
|
"""
|
今日买入代码管理类
|
"""
|
__db = 0
|
redisManager = redis_manager.RedisManager(0)
|
|
def __init__(self):
|
# 挂单中的代码
|
self.delegating_codes_info = {}
|
self.deal_codes = set()
|
self.__load_data()
|
|
@classmethod
|
def __get_redis(cls):
|
return cls.redisManager.getRedis()
|
|
def __load_data(self):
|
"""
|
加载数据
|
:return:
|
"""
|
codes = RedisUtils.smembers(self.__get_redis(), "buy_deal_codes")
|
if codes:
|
self.deal_codes = set(codes)
|
|
def add_deal_code(self, code, order_ref=None):
|
"""
|
添加买入成交的代码
|
:param order_ref:
|
:param code:
|
:return:
|
"""
|
if code in self.deal_codes:
|
return
|
async_log_util.info(logger_trade, f"买入成交：{code}")
|
self.deal_codes.add(code)
|
if order_ref and order_ref in self.delegating_codes_info:
|
del self.delegating_codes_info[order_ref]
|
RedisUtils.sadd_async(self.__db, "buy_deal_codes", code)
|
RedisUtils.expire_async(self.__db, "buy_deal_codes", tool.get_expire())
|
|
def place_order(self, code, order_ref):
|
"""
|
下单
|
:param code: 代码
|
:param order_ref: 索引
|
:return:
|
"""
|
async_log_util.info(logger_trade, f"下单：{code}-{order_ref}")
|
self.delegating_codes_info[order_ref] = code
|
|
def get_buy_codes(self):
|
"""
|
获取买入的代码：成交代码+委托
|
:return:
|
"""
|
codes = set()
|
if self.deal_codes:
|
codes |= self.deal_codes
|
if self.delegating_codes_info:
|
codes |= set(self.delegating_codes_info.values())
|
return codes
|
|
def buy_fail(self, order_ref):
|
"""
|
买入失败
|
:param order_ref:
|
:return:
|
"""
|
async_log_util.info(logger_trade, f"下单失败：{order_ref}")
|
if order_ref in self.delegating_codes_info:
|
del self.delegating_codes_info[order_ref]
|
|
|
if __name__ == '__main__':
|
# 测试代码
|
print(TodayBuyCodeManager().get_buy_codes())
|
