""" 可转债入口函数 """ import datetime import json import logging import multiprocessing import threading import time from code_attribute import target_codes_manager, gpcode_manager, code_market_manager, history_k_data_util from code_attribute.gpcode_manager import CodesNameManager from code_attribute.limit_up_time_manager import FirstLimitUpTimeManager from db.redis_manager_delegate import RedisUtils from huaxin_client import l2_client_for_cb from huaxin_client.client_network import SendResponseSkManager from log_module import async_log_util, log_export from records import huaxin_trade_record_manager from third_data import kpl_data_manager, kpl_util from third_data.kpl_data_manager import PullTask, KPLCodeJXBlockManager, KPLLimitUpDataRecordManager from trade import huaxin_trade_api, huaxin_trade_data_update, huaxin_sell_util, backtest_trade, buy_strategy, \ sell_strategy from trade.buy_strategy import BuyStrategyDataManager, StrategyBuyOrderRefManager from trade.trade_manager import CodeTradeStateManager from trade.trade_settings import WantBuyCodesManager, TradeStateManager from utils import middle_api_protocol, outside_api_command_manager, constant, tool, huaxin_util, socket_util, sell_util, \ output_util, l2_huaxin_util, output_data_util middle_api_protocol.SERVER_PORT = 10008 middle_api_protocol.SERVER_HOST = "43.138.167.68" constant.LOG_DIR = "logs_cb" from log_module.log import logger_debug, logger_trade, printlog import concurrent.futures __cancel_sell_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=8) __BuyStrategyDataManager = BuyStrategyDataManager() def __send_response(data_bytes): sk = SendResponseSkManager.create_send_response_sk(addr=middle_api_protocol.SERVER_HOST, port=middle_api_protocol.SERVER_PORT) try: data_bytes = socket_util.load_header(data_bytes) sk.sendall(data_bytes) result, header_str = socket_util.recv_data(sk) result = json.loads(result) if result["code"] != 0: raise Exception(result['msg']) finally: sk.close() def send_response(data, _client_id, _request_id): data_bytes = json.dumps({"type": "response", "data": data, "client_id": _client_id, "request_id": _request_id}).encode('utf-8') for i in range(3): try: __send_response(data_bytes) printlog("发送数据成功") break except Exception as e1: logging.exception(e1) # 撤长期没有成交的单 def __cancel_not_deal_order(code, order_ref, timeout=3): time.sleep(timeout) # 撤买单 huaxin_trade_api.cancel_order(1, code, "", orderRef=order_ref) def command_callback(client_id, request_id, data): """ 命令回调 :param client_id: :param request_id: :param data: json格式数据 :return: """ type_ = data.get('type') if type_ == outside_api_command_manager.API_TYPE_TRADE: try: trade_type = data["trade_type"] if trade_type == outside_api_command_manager.TRADE_TYPE_ORDER: code = data["code"] direction = data["direction"] volume = data["volume"] price_type = data["price_type"] price = data["price"] sinfo = data["sinfo"] if direction == 2: # price_type: 0-价格笼子 1-跌停价 2-涨停价 3-现价 4-买5价 async_log_util.info(logger_trade, f"API卖: 接收数据-{data}") current_price = None market_info = code_market_manager.get_market_info(code) if market_info: current_price = market_info.price else: price_results = history_k_data_util.HistoryKDatasUtils.get_gp_current_info([code]) current_price = price_results[0]["price"] limit_down_price = target_codes_manager.get_limit_down_price(code) limit_up_price = target_codes_manager.get_limit_up_price(code) order_ref = huaxin_util.create_order_ref() try: result = huaxin_sell_util.start_sell(code, volume, price_type, limit_up_price, limit_down_price, current_price, blocking=True, request_id=request_id, order_ref=order_ref) async_log_util.info(logger_trade, f"API卖结果: {result}") send_response(result, client_id, request_id) except Exception as e: if str(e).find("超时") >= 0: send_response({"code": 0, "data": {"orderRef": order_ref}}, client_id, request_id) else: raise e finally: huaxin_trade_data_update.add_position_list() huaxin_trade_data_update.add_money_list() huaxin_trade_data_update.add_deal_list() else: if not price: limit_down_price = target_codes_manager.get_limit_down_price(code) limit_up_price = target_codes_manager.get_limit_up_price(code) price = sell_util.get_sell_price(price_type, limit_up_price, limit_down_price, None) if not price: raise Exception("尚未获取到买入价格") # 获取买1金额 price = round(float(price), 3) order_ref = huaxin_util.create_order_ref() result = huaxin_trade_api.order(direction, code, volume, price, sinfo=sinfo, order_ref=order_ref, blocking=True, request_id=request_id) # 2s内没成交就撤单 __cancel_sell_thread_pool.submit(__cancel_not_deal_order, code, order_ref) else: result = huaxin_trade_api.order(direction, code, volume, price, sinfo=sinfo, blocking=True, request_id=request_id) huaxin_trade_data_update.add_position_list() huaxin_trade_data_update.add_money_list() huaxin_trade_data_update.add_deal_list() send_response({"code": 0, "data": result}, client_id, request_id) except Exception as e: logger_debug.exception(e) send_response({"code": 1, "msg": str(e)}, client_id, request_id) elif type_ == "get_code_position_info": try: # 查询此仓 code = data.get("code") if code: results = huaxin_trade_record_manager.PositionManager().list_by_day(tool.get_now_date_str("%Y%m%d"), code) else: results = huaxin_trade_record_manager.PositionManager().get_from_cache() if constant.backtest_mode_info: results.clear() position_dict = backtest_trade.position_dict for p in position_dict: translation = position_dict[p] underlying_code = translation["SecurityID"] # TODO 获取当时的价格 translation["buy_list"] = [{"price": str(translation['TradePrice']), "tradeTime": l2_huaxin_util.convert_time(translation['OrderTime']), "volume": 10}] cb_code = target_codes_manager.get_cb_code(underlying_code) underlying_market = code_market_manager.get_market_info(underlying_code) results.append( {"securityID": cb_code, "securityName": gpcode_manager.CodesNameManager().get_code_name(cb_code), "buy_list": translation["buy_list"], "sell_list": [], "currentPosition": 10, "id": cb_code}) for r in results: cb_code = r["securityID"] underlying_code = target_codes_manager.get_underlying_code(cb_code) cb_market = code_market_manager.get_market_info(cb_code) underlying_market = code_market_manager.get_market_info(underlying_code) if cb_market: r["marketInfo"] = {"code": cb_market.code, "name": r["securityName"], "rate": f"{'+' if cb_market.rate > 0.0001 else ''}{round(cb_market.rate * 100, 2)}%", "price": cb_market.price, "lastVolume": cb_market.total_bid_volume // 100, "buy1Money": output_util.money_desc( cb_market.buy1_price * cb_market.buy1_volume), "preClosePrice": cb_market.pre_close_price} r["createTime"] = 0 if underlying_market: if not gpcode_manager.CodesNameManager().get_code_name(underlying_market.code): # 异步请求名称 threading.Thread( target=lambda: gpcode_manager.CodesNameManager().request_code_name(underlying_market.code), daemon=True).start() r["underlyingMarketInfo"] = {"code": underlying_market.code, "name": gpcode_manager.CodesNameManager().get_code_name( underlying_market.code), "rate": f"{'+' if underlying_market.rate > 0.0001 else ''}{round(underlying_market.rate * 100, 2)}%", "price": underlying_market.price, "lastVolume": underlying_market.total_bid_volume // 100, "buy1Money": output_util.money_desc( underlying_market.buy1_price * underlying_market.buy1_volume), "preClosePrice": underlying_market.pre_close_price} r["underlyingDetailInfo"] = output_data_util.load_code_detail_info(underlying_market.code) r["underlyingDetailInfo"]["price"] = underlying_market.price # 获取买点与卖点 if "buy_list" not in r: buys = huaxin_trade_record_manager.DealRecordManager().list_buy_by_code_cache(cb_code) # 根据orderSystemId聚合 temp_dict = {} for b in buys: orderSysID = b["orderSysID"] if orderSysID not in temp_dict: temp_dict[orderSysID] = [] temp_dict[orderSysID].append(b) r["buy_list"] = [] for k in temp_dict: volume = sum([x["volume"] for x in temp_dict[k]]) x = temp_dict[k][0] r["buy_list"].append({"price": str(x["price"]), "tradeTime": x["tradeTime"], "volume": volume, "type": StrategyBuyOrderRefManager().get_strategy_type(x["orderRef"])}) if buys: r["createTime"] = int(buys[0]["tradeTime"].replace(":", "")) if "sell_list" not in r: sells = huaxin_trade_record_manager.DealRecordManager().list_sell_by_code_cache(cb_code) temp_dict = {} for s in sells: orderSysID = s["orderSysID"] if orderSysID not in temp_dict: temp_dict[orderSysID] = [] temp_dict[orderSysID].append(s) r["sell_list"] = [] for k in temp_dict: volume = sum([x["volume"] for x in temp_dict[k]]) x = temp_dict[k][0] r["sell_list"].append({"price": str(x["price"]), "tradeTime": x["tradeTime"], "volume": volume}) send_response({"code": 0, "data": results}, client_id, request_id) except Exception as e: logger_debug.exception(e) send_response({"code": 1, "msg": str(e)}, client_id, request_id) elif type_ == "refresh_trade_data": # 刷新交易数据 ctype = data.get("ctype") if ctype == "money": huaxin_trade_data_update.add_money_list() elif ctype == "position_list": huaxin_trade_data_update.add_position_list() elif ctype == "deal_list": huaxin_trade_data_update.add_deal_list() elif ctype == "delegate_list": huaxin_trade_data_update.add_delegate_list("手动刷新") send_response({"code": 0, "data": {}}, client_id, request_id) elif type_ == outside_api_command_manager.API_TYPE_COMMON_REQUEST: # 常规接口 ctype = data['ctype'] if ctype == 'get_account_money': # 获取账户资金 result = huaxin_trade_record_manager.MoneyManager.get_data() send_response({"code": 0, "data": result}, client_id, request_id) elif ctype == 'set_backtest_mode': try: # 设置回测模式 date = data.get("date") mode = data.get("mode") if not date: date = tool.get_now_date_str() if mode: # 开始回撤 constant.backtest_mode_info = (True, date) threading.Thread(target=lambda: backtest_trade.start_backtest(date), daemon=True).start() else: # 结束回撤 constant.backtest_mode_info = None send_response({"code": 0, "data": {}}, client_id, request_id) except Exception as e: send_response({"code": 0, "data": {}, "msg": str(e)}, client_id, request_id) elif ctype == "want_buy_codes": try: operate = data["operate"] if operate == outside_api_command_manager.OPERRATE_ADD: code = data["code"] WantBuyCodesManager().add_code(code, buy_strategy.STRATEGY_TYPE_LIMIT_UP) WantBuyCodesManager().add_code(code, buy_strategy.STRATEGY_TYPE_RISE_HIGH_WITH_BLOCKS) send_response({"code": 0, "data": {}}, client_id, request_id) elif operate == outside_api_command_manager.OPERRATE_GET: code_infos = WantBuyCodesManager().list_code() codes = set([x.split("-")[0] for x in code_infos]) code_infos = [(x, CodesNameManager.get_code_name(x)) for x in codes] send_response({"code": 0, "data": code_infos}, client_id, request_id) elif operate == outside_api_command_manager.OPERRATE_DELETE: code = data["code"] WantBuyCodesManager().remove_code(code, buy_strategy.STRATEGY_TYPE_LIMIT_UP) WantBuyCodesManager().remove_code(code, buy_strategy.STRATEGY_TYPE_RISE_HIGH_WITH_BLOCKS) send_response({"code": 0, "data": {}}, client_id, request_id) except Exception as e: send_response({"code": 1, "msg": str(e)}, client_id, request_id) elif ctype == "trade_state": try: operate = data["operate"] if operate == outside_api_command_manager.OPERRATE_SET: state = data["state"] if state > 0: TradeStateManager().open_buy() else: TradeStateManager().close_buy() send_response({"code": 0, "data": {}}, client_id, request_id) elif operate == outside_api_command_manager.OPERRATE_GET: state = 1 if TradeStateManager().is_can_buy() else 0 send_response({"code": 0, "data": {"state": state}}, client_id, request_id) except Exception as e: send_response({"code": 1, "msg": str(e)}, client_id, request_id) def test(): time.sleep(5) print("获取资金：", huaxin_trade_api.get_money()) # print("获取成交：", huaxin_trade_api.get_deal_list()) # print("下单：", huaxin_trade_api.order(1, "127075", 10, 140.5, blocking=True)) def __get_buy_money(strategy_type): time_str = tool.get_now_time_str().replace(":", "") if int(time_str) < int("103000"): money = 40000 elif int(time_str) < int("113000"): money = 30000 elif int(time_str) < int("140000"): money = 20000 else: money = 10000 if strategy_type == buy_strategy.STRATEGY_TYPE_RISE_HIGH_WITH_BLOCKS: return int(money * 0.8) elif strategy_type == buy_strategy.STRATEGY_TYPE_LIMIT_UP: return int(money * 0.2) else: return money def read_l2_results(trade_call_back_queue): while True: try: result = trade_call_back_queue.get() if result: # 获取可以买的代码 code, trade_time = result[0], result[1] buy_infos = result[3] for buy_info in buy_infos: if buy_info[0]: if not TradeStateManager().is_can_buy(): continue # 策略类型 strategy_type = buy_info[1] # 获取股票代码的可转债代码 cb_code = target_codes_manager.get_cb_code(code) if CodeTradeStateManager().get_trade_state(cb_code, strategy_type) == CodeTradeStateManager.TRADE_STATE_ALREADY_BUY and not WantBuyCodesManager().is_in_cache( cb_code, strategy_type): # 已经买了且没在想买单 continue # 如果之前就涨停了且没在想买单中 underlying_code = target_codes_manager.get_underlying_code(cb_code) underlying_limit_up_time = FirstLimitUpTimeManager().get_first_limit_up_time(underlying_code) if not WantBuyCodesManager().is_in_cache(cb_code, strategy_type) and underlying_limit_up_time: continue # 获取可转债的涨停价 market_info = code_market_manager.get_market_info(cb_code) limit_up_price = target_codes_manager.get_limit_up_price(cb_code) if market_info: if market_info.rate > 0.139: async_log_util.info(logger_trade, f"可转债涨幅过高：：{cb_code}-{market_info.rate}") continue volume = int(__get_buy_money(strategy_type) / float(limit_up_price)) volume = (volume // 10) * 10 buy_price = round(min(float(market_info.price * 1.02), float(limit_up_price)), 3) async_log_util.info(logger_trade, f"准备下单：{cb_code}-{buy_price}-{buy_info}") # 买入20股 result = huaxin_trade_api.order(1, cb_code, volume, buy_price, blocking=True) if type(result) == dict and result['code'] == 0: orderRef = result['data']['orderRef'] StrategyBuyOrderRefManager().add(orderRef, strategy_type) CodeTradeStateManager().set_trade_state(cb_code, strategy_type, CodeTradeStateManager.TRADE_STATE_ALREADY_BUY) # 移除想买单 WantBuyCodesManager().remove_code(cb_code, strategy_type) async_log_util.info(logger_trade, f"可转债下单结果：{result}") huaxin_trade_data_update.add_position_list() huaxin_trade_data_update.add_money_list() huaxin_trade_data_update.add_deal_list() except Exception as e: logger_debug.exception(e) time.sleep(1) finally: pass def __read_market_data(queue_market: multiprocessing.Queue): while True: try: result = queue_market.get() if result: # (代码, 最近的价格, 涨幅, 买1价， 买1量, 成交总量, 买入量, 卖出量, 昨日收盘价, 时间戳) code_market_manager.set_market_info(result) try: time_str = l2_huaxin_util.convert_time(result[9]) if int(time_str.replace(":", "")) < int("092455"): continue except Exception as e: pass code = result[0] # 正股需要加载板块 if tool.is_stock(code): # 正股代码 limit_up_price = tool.get_limit_up_price(code, result[8]) # 涨幅大于5%才开始获取板块 if result[2] > 0.05: KPLCodeJXBlockManager().load_jx_blocks(result[0], result[3], float(limit_up_price), KPLLimitUpDataRecordManager.get_current_reasons()) FirstLimitUpTimeManager().process(result) else: # 可转债代码,处理 sell_strategy.process_market_info(result, __BuyStrategyDataManager) except Exception as e: logger_debug.exception(e) time.sleep(1) finally: pass def __init_data(): """ 初始化参数 :return: """ try: market_dict = log_export.load_latest_market_info() for k in market_dict: code_market_manager.set_market_info(market_dict[k], with_log=False) except Exception as e: logger_debug.exception(e) def __kpl_limit_up_callback(results): if results: result_list = kpl_util.parseLimitUpData(results) kpl_data_manager.KPLLimitUpDataRecordManager.save_record(tool.get_now_date_str(), result_list) if __name__ == '__main__': # ===========初始化数据========== try: target_codes_manager.load_valid_codes_info() except Exception as e: logger_debug.exception(e) __init_data() trade_call_back_queue = multiprocessing.Queue() # 华鑫交易数据更新 huaxin_trade_data_update.run() # 定时拉取开盘啦涨停数据 threading.Thread(target=lambda: PullTask.run_limit_up_task(__kpl_limit_up_callback), daemon=True).start() # 仿真交易不运行交易客户端 # # ===========运行交易外部API========== # # # # # 策略与交易通信队列 # # 交易结果读取, 交易命令队列与交易查询队列设置为同一个 # queue_strategy_r_trade_w, queue_strategy_w_trade_r, queue_strategy_w_trade_r_for_query = multiprocessing.Queue(), multiprocessing.Queue(), multiprocessing.Queue() # huaxin_trade_api.run_trade(queue_strategy_r_trade_w, queue_strategy_w_trade_r, queue_strategy_w_trade_r_for_query) # # # # ===========运行交易端========== # tradeProcess = multiprocessing.Process( # target=trade_client_for_cb.run, # args=(queue_strategy_w_trade_r, queue_strategy_w_trade_r_for_query, queue_strategy_r_trade_w,)) # tradeProcess.start() # ===========运行本地API接口========== # middle_api_protocol.SERVER_HOST = "192.168.3.122" manager = outside_api_command_manager.NewApiCommandManager() manager.init(middle_api_protocol.SERVER_HOST, middle_api_protocol.SERVER_PORT, command_callback, [("trade_cb", 50)]) manager.run(blocking=False) # threading.Thread(target=test, daemon=True).start() # ===========读取根据L2制定的买入策略========== threading.Thread(target=read_l2_results, args=(trade_call_back_queue,), daemon=True).start() # ===========异步日志持久化========== threading.Thread(target=async_log_util.run_sync, daemon=True).start() # ===========Redis缓存====================== threading.Thread(target=RedisUtils.run_loop, daemon=True).start() # 运行L2数据监听队列 queue_market = multiprocessing.Queue() threading.Thread(target=__read_market_data, args=(queue_market,), daemon=True).start() l2_client_for_cb.run(trade_call_back_queue, queue_market)