diff --git a/api/qmt_api_new.py b/api/qmt_api_new.py new file mode 100644 index 0000000..381fbda --- /dev/null +++ b/api/qmt_api_new.py @@ -0,0 +1,1534 @@ +# -*- coding: gbk -*- +import json +import locale +import os +import datetime +from pathlib import Path +from tornado.web import Application, RequestHandler, HTTPError +from tornado.ioloop import IOLoop +import logging + +# Configuration +ACCOUNT_ID = os.environ.get('QMT_ACCOUNT_ID', '') +DATA_DIR = os.environ.get('QMT_DATA_DIR', 'D:\qmt_strategy_data') +TOKEN="QMTbyYanweidong" +PORT = 10086 + +# =================================== +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) +locale.setlocale(locale.LC_CTYPE, 'chinese') + + +def safe_call(func, *args, **kwargs): + try: + return func(*args, **kwargs) + except HTTPError: + raise + except Exception as e: + logger.exception("%s call failed", func.__name__) + raise HTTPError( + 502, + reason="QMT upstream call failed: %s" % func.__name__, + ) from e + + +# ============= BaseHandler ============= + +AUTH_EXEMPT = set() + + +def no_auth(cls): + AUTH_EXEMPT.add(cls) + return cls + + +class BaseHandler(RequestHandler): + def prepare(self): + if self.__class__ not in AUTH_EXEMPT: + token = self.request.headers.get('X-Token') + if token != TOKEN: + raise HTTPError(401, "Authentication failed: invalid or missing token") + + def set_default_headers(self): + self.set_header("Content-Type", "application/json; charset=utf-8") + + def write_error(self, status_code, **kwargs): + self.finish(json.dumps({ + "error": self._reason, + "status_code": status_code + }, ensure_ascii=False)) + + def ctx(self): + return self.application.ContextInfo + + def acc(self): + return self.application.accountID + + +# ============= 1. ContextInfo properties ============= +# ContextInfo.period - Get the current period +class ContextPeriodHandler(BaseHandler): + def get(self): + self.write(json.dumps({"period": self.ctx().period}, ensure_ascii=False)) + +# ContextInfo.barpos - Get the current bar index +class ContextBarposHandler(BaseHandler): + def get(self): + self.write(json.dumps({"barpos": self.ctx().barpos}, ensure_ascii=False)) + +# ContextInfo.time_tick_size - Get the current bar count +class ContextTimeTickSizeHandler(BaseHandler): + def get(self): + self.write(json.dumps({"time_tick_size": self.ctx().time_tick_size}, ensure_ascii=False)) + +# ContextInfo.stockcode - Get the current chart symbol +class ContextStockCodeHandler(BaseHandler): + def get(self): + self.write(json.dumps({"stockcode": self.ctx().stockcode}, ensure_ascii=False)) + +# ContextInfo.dividend_type - Get the current adjustment mode +class ContextDividendTypeHandler(BaseHandler): + def get(self): + self.write(json.dumps({"dividend_type": self.ctx().dividend_type}, ensure_ascii=False)) + +# ContextInfo.market - Get the current chart market +class ContextMarketHandler(BaseHandler): + def get(self): + self.write(json.dumps({"market": self.ctx().market}, ensure_ascii=False)) + +# ContextInfo.do_back_test - Check whether backtesting is enabled +class ContextDoBackTestHandler(BaseHandler): + def get(self): + self.write(json.dumps({"do_back_test": self.ctx().do_back_test}, ensure_ascii=False)) + +# ContextInfo.benchmark - Get the backtest benchmark +class ContextBenchmarkHandler(BaseHandler): + def get(self): + self.write(json.dumps({"benchmark": self.ctx().benchmark}, ensure_ascii=False)) + +# ContextInfo.capital - Get the initial backtest capital +class ContextCapitalHandler(BaseHandler): + def get(self): + self.write(json.dumps({"capital": self.ctx().capital}, ensure_ascii=False)) + +# ContextInfo.get_universe() - Get symbols in the universe +class ContextUniverseHandler(BaseHandler): + def get(self): + self.write(json.dumps({"universe": self.ctx().get_universe()}, ensure_ascii=False)) + + +# ============= 2. Data queries (ContextInfo get_*) ============= +# ContextInfo.get_stock_name() - Get a stock name by symbol +class StockNameHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(self.ctx().get_stock_name, stockcode) + self.write(json.dumps({"stockcode": stockcode, "name": ret}, ensure_ascii=False)) + +# get_open_date() - Get the listing date by symbol +class OpenDateHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(get_open_date, stockcode) + self.write(json.dumps({"stockcode": stockcode, "open_date": ret}, ensure_ascii=False)) + +# ContextInfo.get_last_volume() - Get the latest outstanding shares +class LastVolumeHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(self.ctx().get_last_volume, stockcode) + if ret is None: + raise HTTPError(500, "Failed to get outstanding shares") + self.write(json.dumps({"stockcode": stockcode, "last_volume": ret}, ensure_ascii=False)) + +# ContextInfo.get_bar_timetag() - Get the bar timestamp +class BarTimetagHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + index = int(data.get('index', -1)) + ret = safe_call(self.ctx().get_bar_timetag, index) + self.write(json.dumps({"index": index, "timetag": ret}, ensure_ascii=False)) + +# ContextInfo.get_tick_timetag() - Get the latest tick timestamp +class TickTimetagHandler(BaseHandler): + def get(self): + ret = safe_call(self.ctx().get_tick_timetag) + self.write(json.dumps({"timetag": ret}, ensure_ascii=False)) + +# ContextInfo.get_sector() - Get index constituents +class SectorHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + sector = data.get('sector', '') + realtime = data.get('realtime', '0') + if not sector: + raise HTTPError(400, "need args sector") + ret = safe_call(self.ctx().get_sector, sector, int(realtime) if realtime != '0' else 0) + self.write(json.dumps({"sector": sector, "stocks": ret or []}, ensure_ascii=False)) + +# ContextInfo.get_industry() - Get industry constituents +class IndustryHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + industry = data.get('industry', '') + if not industry: + raise HTTPError(400, "need args industry") + print(industry) + ret = safe_call(self.ctx().get_industry, industry) + self.write(json.dumps({"industry": industry, "stocks": ret or []}, ensure_ascii=False)) + +# ContextInfo.get_stock_list_in_sector() - Get sector constituents +class StockListInSectorHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + sectorname = data.get('sectorname', '') + if not sectorname: + raise HTTPError(400, "need args sectorname") + ret = safe_call(self.ctx().get_stock_list_in_sector, sectorname) + self.write(json.dumps({"sectorname": sectorname, "stocks": ret or []}, ensure_ascii=False)) + +# ContextInfo.get_weight_in_index() - Get weights in an index +class WeightInIndexHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + indexcode = data.get('indexcode', '') + stockcode = data.get('stockcode', '') + ret = safe_call(self.ctx().get_weight_in_index, indexcode, stockcode) + self.write(json.dumps({"indexcode": indexcode, "stockcode": stockcode, "weight": ret}, ensure_ascii=False)) + +# ContextInfo.get_contract_multiplier() - Get the contract multiplier +class ContractMultiplierHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + contractcode = data.get('contractcode', '') + ret = safe_call(self.ctx().get_contract_multiplier, contractcode) + self.write(json.dumps({"contractcode": contractcode, "multiplier": ret}, ensure_ascii=False)) + +# ContextInfo.get_risk_free_rate() - Get the risk-free rate +class RiskFreeRateHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + index = int(data.get('index', '-1')) + ret = safe_call(self.ctx().get_risk_free_rate, index) + self.write(json.dumps({"index": index, "risk_free_rate": ret}, ensure_ascii=False)) + +# ContextInfo.get_date_location() - Get the bar index for a date +class DateLocationHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + strdate = data.get('strdate', '') + ret = safe_call(self.ctx().get_date_location, strdate) + self.write(json.dumps({"strdate": strdate, "location": ret}, ensure_ascii=False)) + +# ContextInfo.get_history_data() - Get historical market data for multiple symbols +class HistoryDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + length = int(data.get('len', '10')) + period = data.get('period', '1d') + field = data.get('field', 'close') + dividend_type = int(data.get('dividend_type', '0')) + skip_paused = data.get('skip_paused', 'true').lower() == 'true' + ret = safe_call(self.ctx().get_history_data, length, period, field, dividend_type, skip_paused) + self.write(json.dumps({"data": ret} if ret else {"error": "Failed to get historical data"}, ensure_ascii=False)) + +# ContextInfo.get_market_data() - Get market data as a DataFrame +class MarketDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + fields = data.get('fields', '') + stock_code = data.get('stock_code', '') + start_time = data.get('start_time', '') + end_time = data.get('end_time', '') + period = data.get('period', '1d') + dividend_type = data.get('dividend_type', 'none') + count = int(data.get('count', '-1')) + fields_list = [f.strip() for f in fields.split(',')] if fields else [] + stock_list = [s.strip() for s in stock_code.split(',')] if stock_code else [] + ret = safe_call(self.ctx().get_market_data, fields_list, stock_list, start_time, end_time, True, period, dividend_type, count) + if ret is None: + raise HTTPError(500, "Failed to get market data") + if hasattr(ret, 'to_dict'): + ret = ret.to_dict() + self.write(json.dumps({"data": ret}, ensure_ascii=False, default=str)) + +# ContextInfo.get_market_data_ex() - Get extended Level 2 market data +class MarketDataExHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + fields = data.get('fields', '') + stock_code = data.get('stock_code', '') + period = data.get('period', 'follow') + start_time = data.get('start_time', '') + end_time = data.get('end_time', '') + count = int(data.get('count', '-1')) + dividend_type = data.get('dividend_type', 'follow') + fields_list = [f.strip() for f in fields.split(',')] if fields else [] + stock_list = [s.strip() for s in stock_code.split(',')] if stock_code else [] + ret = safe_call(self.ctx().get_market_data_ex, fields_list, stock_list, period, start_time, end_time, count, dividend_type) + if ret is None: + raise HTTPError(500, "Failed to get extended market data") + result = {} + for k, v in ret.items(): + if hasattr(v, 'to_dict'): + result[k] = v.to_dict() + else: + result[k] = str(v) + self.write(json.dumps({"data": result}, ensure_ascii=False, default=str)) + +# ContextInfo.get_full_tick() - Get full tick data +class FullTickHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stocks = data.get('stocks', []) + #if not stocks: + # raise HTTPError(400, "need args stocks") + ret = safe_call(self.ctx().get_full_tick, stocks) + if not ret: + raise HTTPError(500, "Failed to get tick data") + self.write(json.dumps(ret, ensure_ascii=False, default=str)) + +# ContextInfo.get_divid_factors() - Get dividend and adjustment factors +class DividFactorsHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(self.ctx().get_divid_factors, stockcode) + self.write(json.dumps({"stockcode": stockcode, "factors": ret or {}}, ensure_ascii=False)) + +# ContextInfo.get_main_contract() - Get the main futures contract +class MainContractHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + codemarket = data.get('codemarket', '') + ret = safe_call(self.ctx().get_main_contract, codemarket) + self.write(json.dumps({"codemarket": codemarket, "main_contract": ret}, ensure_ascii=False)) + +# timetag_to_datetime() - Convert a millisecond timestamp to datetime +class TimetagToDatetimeHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + timetag = int(data.get('timetag', '0')) + fmt = data.get('format', '%Y-%m-%d %H:%M:%S') + ret = safe_call(timetag_to_datetime, timetag, fmt) + self.write(json.dumps({"timetag": timetag, "datetime": ret}, ensure_ascii=False)) + +# ContextInfo.get_total_share() - Get total shares +class TotalShareHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(self.ctx().get_total_share, stockcode) + self.write(json.dumps({"stockcode": stockcode, "total_share": ret}, ensure_ascii=False)) + +# ContextInfo.get_trading_dates() - Get the trading-day list +class TradingDatesHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + start_date = data.get('start_date', '') + end_date = data.get('end_date', '') + count = data.get('count', '') + period = data.get('period', '1d') + count_int = int(count) if count else -1 + ret = safe_call(self.ctx().get_trading_dates, stockcode, start_date, end_date, count_int, period) + self.write(json.dumps({"dates": ret or []}, ensure_ascii=False)) + +# ContextInfo.get_svol() - Get sell-side volume +class SvolHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(self.ctx().get_svol, stockcode) + self.write(json.dumps({"stockcode": stockcode, "svol": ret}, ensure_ascii=False)) + +# ContextInfo.get_bvol() - Get buy-side volume +class BvolHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(self.ctx().get_bvol, stockcode) + self.write(json.dumps({"stockcode": stockcode, "bvol": ret}, ensure_ascii=False)) + +# ContextInfo.get_longhubang() - Get Dragon-Tiger List data +class LonghubangHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stock_list = data.get('stock_list', '') + startTime = data.get('startTime', '') + endTime = data.get('endTime', '') + slist = [s.strip() for s in stock_list.split(',')] if stock_list else [] + ret = safe_call(self.ctx().get_longhubang, slist, startTime, endTime) + if hasattr(ret, 'to_dict'): + ret = ret.to_dict() + self.write(json.dumps({"data": ret} if ret else {"error": "Failed to get Dragon-Tiger List data"}, ensure_ascii=False, default=str)) + +# get_top10_share_holder() - Get top-ten shareholder data +class Top10ShareHolderHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stock_list = data.get('stock_list', '') + data_name = data.get('data_name', 'holder') + start_time = data.get('start_time', '') + end_time = data.get('end_time', '') + slist = [s.strip() for s in stock_list.split(',')] if stock_list else [] + ret = safe_call(get_top10_share_holder, slist, data_name, start_time, end_time) + if hasattr(ret, 'to_dict'): + ret = ret.to_dict() + self.write(json.dumps({"data": ret} if ret else {"error": "Failed to get top-ten shareholder data"}, ensure_ascii=False, default=str)) + +# ContextInfo.get_option_detail_data() - Get option details +class OptionDetailHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + optioncode = data.get('optioncode', '') + ret = safe_call(self.ctx().get_option_detail_data, optioncode) + self.write(json.dumps({"optioncode": optioncode, "detail": ret or {}}, ensure_ascii=False)) + +# ContextInfo.get_turnover_rate() - Get turnover rate +class TurnoverRateHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stock_list = data.get('stock_list', '') + startTime = data.get('startTime', '') + endTime = data.get('endTime', '') + slist = [s.strip() for s in stock_list.split(',')] if stock_list else [] + ret = safe_call(self.ctx().get_turnover_rate, slist, startTime, endTime) + if hasattr(ret, 'to_dict'): + ret = ret.to_dict() + self.write(json.dumps({"data": ret} if ret else {"error": "Failed to get turnover rate"}, ensure_ascii=False, default=str)) + +# get_etf_info() - Get ETF creation/redemption and constituent data +class EtfInfoHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(get_etf_info, stockcode) + self.write(json.dumps({"stockcode": stockcode, "info": ret or {}}, ensure_ascii=False, default=str)) + +# get_etf_iopv() - Get the ETF indicative optimized portfolio value +class EtfIopvHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(get_etf_iopv, stockcode) + self.write(json.dumps({"stockcode": stockcode, "iopv": ret}, ensure_ascii=False)) + +# ContextInfo.get_instrumentdetail() - Get instrument details +class InstrumentDetailHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(self.ctx().get_instrumentdetail, stockcode) + self.write(json.dumps({"stockcode": stockcode, "detail": ret or {}}, ensure_ascii=False, default=str)) + +# ContextInfo.get_contract_expire_date() - Get the futures contract expiration date +class ContractExpireDateHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + codemarket = data.get('codemarket', '') + ret = safe_call(self.ctx().get_contract_expire_date, codemarket) + self.write(json.dumps({"codemarket": codemarket, "expire_date": ret}, ensure_ascii=False)) + +# ContextInfo.get_option_undl_data() - Get option products for an underlying asset +class OptionUndlDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + undl_code_ref = data.get('undl_code_ref', '') + ret = safe_call(self.ctx().get_option_undl_data, undl_code_ref) + self.write(json.dumps({"data": ret or []}, ensure_ascii=False, default=str)) + +# ContextInfo.get_financial_data() - Get financial data +class FinancialDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + tabname = data.get('tabname', '') + colname = data.get('colname', '') + market = data.get('market', '') + code = data.get('code', '') + report_type = data.get('report_type', 'report_time') + barpos = int(data.get('barpos', '-1')) + if tabname and colname and market and code: + ret = safe_call(self.ctx().get_financial_data, tabname, colname, market, code, report_type, barpos) + else: + field_list = data.get('fieldList', '') + stock_list = data.get('stockList', '') + start_date = data.get('startDate', '') + end_date = data.get('endDate', '') + fields = [f.strip() for f in field_list.split(',')] if field_list else [] + stocks = [s.strip() for s in stock_list.split(',')] if stock_list else [] + rtype = data.get('report_type', 'announce_time') + ret = safe_call(self.ctx().get_financial_data, fields, stocks, start_date, end_date, rtype) + if hasattr(ret, 'to_dict'): + ret = ret.to_dict() + self.write(json.dumps({"data": ret} if ret is not None else {"error": "Failed to get financial data"}, ensure_ascii=False, default=str)) + +# ContextInfo.get_factor_data() - Get multi-factor data +class FactorDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + field_list = data.get('fieldList', '') + stock_list = data.get('stockList', '') + stock_code = data.get('stockCode', '') + start_date = data.get('startDate', '') + end_date = data.get('endDate', '') + fields = [f.strip() for f in field_list.split(',')] if field_list else [] + if stock_code: + ret = safe_call(self.ctx().get_factor_data, fields, stock_code, start_date, end_date) + else: + stocks = [s.strip() for s in stock_list.split(',')] if stock_list else [] + ret = safe_call(self.ctx().get_factor_data, fields, stocks, start_date, end_date) + if hasattr(ret, 'to_dict'): + ret = ret.to_dict() + self.write(json.dumps({"data": ret} if ret is not None else {"error": "Failed to get factor data"}, ensure_ascii=False, default=str)) + +# ContextInfo.get_his_st_data() - Get historical ST data +class HisStDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockCode = data.get('stockCode', '') + ret = safe_call(self.ctx().get_his_st_data, stockCode) + self.write(json.dumps({"stockCode": stockCode, "data": ret or {}}, ensure_ascii=False)) + +# ContextInfo.get_his_index_data() - Get historical index data +class HisIndexDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + index = data.get('index', '') + ret = safe_call(self.ctx().get_his_index_data, index) + self.write(json.dumps({"index": index, "data": ret or {}}, ensure_ascii=False, default=str)) + +# ContextInfo.get_all_subscription() - Get all current market-data subscriptions +class AllSubscriptionHandler(BaseHandler): + def get(self): + ret = safe_call(self.ctx().get_all_subscription) + self.write(json.dumps({"subscriptions": ret or {}}, ensure_ascii=False, default=str)) + +# ContextInfo.get_option_list() - Get a specified option list +class OptionListHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + undl_code = data.get('undl_code', '') + dedate = data.get('dedate', '') + opttype = data.get('opttype', '') + isavailable = data.get('isavailable', 'true').lower() == 'true' + ret = safe_call(self.ctx().get_option_list, undl_code, dedate, opttype, isavailable) + self.write(json.dumps({"option_list": ret or []}, ensure_ascii=False)) + +# ContextInfo.get_his_contract_list() - Get expired contracts +class HisContractListHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + market = data.get('market', '') + ret = safe_call(self.ctx().get_his_contract_list, market) + self.write(json.dumps({"market": market, "contracts": ret or []}, ensure_ascii=False)) + +# ContextInfo.get_option_iv() - Get real-time option implied volatility +class OptionIvHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + optioncode = data.get('optioncode', '') + ret = safe_call(self.ctx().get_option_iv, optioncode) + self.write(json.dumps({"optioncode": optioncode, "iv": ret}, ensure_ascii=False)) + +# ContextInfo.bsm_price() - Calculate a European option price with the Black-Scholes model +class BsmPriceHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + optionType = data.get('optionType', 'C') + objectPrices = data.get('objectPrices', '') + strikePrice = float(data.get('strikePrice', '0')) + riskFree = float(data.get('riskFree', '0')) + sigma = float(data.get('sigma', '0')) + days = int(data.get('days', '0')) + dividend = float(data.get('dividend', '0')) + try: + op = float(objectPrices) + except ValueError: + op = [float(x) for x in objectPrices.split(',')] + ret = safe_call(self.ctx().bsm_price, optionType, op, strikePrice, riskFree, sigma, days, dividend) + self.write(json.dumps({"price": ret}, ensure_ascii=False, default=str)) + +# ContextInfo.bsm_iv() - Calculate European option implied volatility +class BsmIvHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + optionType = data.get('optionType', 'C') + objectPrices = float(data.get('objectPrices', '0')) + strikePrice = float(data.get('strikePrice', '0')) + optionPrice = float(data.get('optionPrice', '0')) + riskFree = float(data.get('riskFree', '0')) + days = int(data.get('days', '0')) + dividend = float(data.get('dividend', '0')) + ret = safe_call(self.ctx().bsm_iv, optionType, objectPrices, strikePrice, optionPrice, riskFree, days, dividend) + self.write(json.dumps({"iv": ret}, ensure_ascii=False)) + +# ContextInfo.get_local_data() - Get market data from local storage +class LocalDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stock_code = data.get('stock_code', '') + start_time = data.get('start_time', '') + end_time = data.get('end_time', '') + period = data.get('period', '1d') + divid_type = data.get('divid_type', 'none') + count = int(data.get('count', '-1')) + ret = safe_call(self.ctx().get_local_data, stock_code, start_time, end_time, period, divid_type, count) + if ret is None: + raise HTTPError(500, "Failed to get local market data") + self.write(json.dumps({"data": ret}, ensure_ascii=False, default=str)) + +# ContextInfo.subscribe_quote() - Subscribe to market data +class SubscribeQuoteHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stock_code = data.get('stock_code', '') + period = data.get('period', 'follow') + dividend_type = data.get('dividend_type', 'follow') + ret = safe_call(self.ctx().subscribe_quote, stock_code, period, dividend_type) + self.write(json.dumps({"status": "success" if ret is not None else "failed", "sub_id": ret}, ensure_ascii=False)) + +# ContextInfo.unsubscribe_quote() - Unsubscribe from market data +class UnsubscribeQuoteHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + sub_id = int(data.get('sub_id', '0')) + safe_call(self.ctx().unsubscribe_quote, sub_id) + self.write(json.dumps({"status": "success", "sub_id": sub_id}, ensure_ascii=False)) + + +# ============= 3. Predicate functions (is_*) ============= +# ContextInfo.is_last_bar() - Check whether this is the last bar +class IsLastBarHandler(BaseHandler): + def get(self): + ret = safe_call(self.ctx().is_last_bar) + self.write(json.dumps({"is_last_bar": ret}, ensure_ascii=False)) + +# ContextInfo.is_new_bar() - Check whether this is a new bar +class IsNewBarHandler(BaseHandler): + def get(self): + ret = safe_call(self.ctx().is_new_bar) + self.write(json.dumps({"is_new_bar": ret}, ensure_ascii=False)) + +# ContextInfo.is_suspended_stock() - Check whether a stock is suspended +class IsSuspendedStockHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stockcode = data.get('stockcode', '') + ret = safe_call(self.ctx().is_suspended_stock, stockcode) + self.write(json.dumps({"stockcode": stockcode, "is_suspended": ret}, ensure_ascii=False)) + +# is_sector_stock() - Check whether a stock belongs to a sector +class IsSectorStockHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + sectorname = data.get('sectorname', '') + market = data.get('market', '') + stockcode = data.get('stockcode', '') + ret = safe_call(is_sector_stock, sectorname, market, stockcode) + self.write(json.dumps({"sectorname": sectorname, "stockcode": stockcode, "is_in_sector": ret}, ensure_ascii=False)) + +# is_typed_stock() - Check whether a stock belongs to a category +class IsTypedStockHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + stocktypenum = int(data.get('stocktypenum', '0')) + market = data.get('market', '') + stockcode = data.get('stockcode', '') + ret = safe_call(is_typed_stock, stocktypenum, market, stockcode) + self.write(json.dumps({"stocktypenum": stocktypenum, "stockcode": stockcode, "result": ret}, ensure_ascii=False)) + +# get_industry_name_of_stock() - Get a stock's industry classification +class GetIndustryNameOfStockHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + industryType = data.get('industryType', '') + stockcode = data.get('stockcode', '') + ret = safe_call(get_industry_name_of_stock, industryType, stockcode) + self.write(json.dumps({"industryType": industryType, "stockcode": stockcode, "industry_name": ret}, ensure_ascii=False)) + + +# ============= 4. Trading functions ============= +# passorder() - Submit a general trading order +class PassorderHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + opType = int(data['opType']) + orderType = int(data.get('orderType', 1101)) + stock = data['stock'] + pr_type = int(data.get('prType', 11)) + price = float(data['price']) + volume = int(data['volume']) + quickTrade = int(data.get('quickTrade', 2)) + strategy_name = str(data.get('strategyName', '')).strip() + order_id = str(data.get('orderId', '')).strip() + except (json.JSONDecodeError, KeyError, TypeError, ValueError) as e: + raise HTTPError(400, reason="Invalid order parameters: %s" % e) from e + + # QMT stores strategyName in the order remark; preserve the signal key and local order ID. + # Put the local ID first so restart reconciliation still works if QMT truncates the remark. + remark = '|'.join(part for part in (order_id, strategy_name) if part)[:24] + try: + order_ref = passorder(opType, orderType, self.acc(), stock, pr_type, price, volume, remark, quickTrade, self.ctx()) + except HTTPError: + raise + except Exception as e: + logger.exception("passorder failed") + raise HTTPError(502, reason="QMT order submission failed") from e + + if not order_ref: + raise HTTPError(502, reason="QMT did not return a valid order reference") + + self.write(json.dumps({ + "status": "success", + "opType": opType, + "stock": stock, + "strategy_name": strategy_name, + "local_order_id": order_id, + "order_ref": str(order_ref) + }, ensure_ascii=False)) + +# algo_passorder() - Submit an algorithmic order +class AlgoPassorderHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + order_ref = algo_passorder( + int(data['opType']), int(data.get('orderType', 1101)), + self.acc(), data['stock'], int(data.get('prType', -1)), + float(data['price']), int(data['volume']), + data.get('strategyName', ''), int(data.get('quickTrade', 2)), + data.get('userOrderId', ''), data.get('userOrderParam', {}), + self.ctx() + ) + self.write(json.dumps({"status": "success", "order_ref": str(order_ref) if order_ref else "unknown"}, ensure_ascii=False)) + except Exception as e: + logger.exception("algo_passorder failed") + raise HTTPError(400, f"Algorithmic order submission failed: {str(e)}") + +# smart_algo_passorder() - Submit a smart algorithmic order +class SmartAlgoPassorderHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + order_ref = smart_algo_passorder( + int(data['opType']), int(data.get('orderType', 1101)), + self.acc(), data['stock'], int(data.get('prType', -1)), + float(data['price']), int(data['volume']), + data['smartAlgoType'], int(data.get('limitOverRate', 0)), + int(data.get('minAmountPerOrder', 0)), + data.get('startTime', ''), data.get('endTime', ''), + self.ctx() + ) + self.write(json.dumps({"status": "success", "order_ref": str(order_ref) if order_ref else "unknown"}, ensure_ascii=False)) + except Exception as e: + logger.exception("smart_algo_passorder failed") + raise HTTPError(400, f"Smart algorithmic order submission failed: {str(e)}") + +# order_lots() - Trade a specified number of lots +class OrderLotsHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + order_lots(data['stock'], int(data['lots']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "order_lots", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("order_lots failed") + raise HTTPError(400, f"Order submission failed: {str(e)}") + +# order_value() - Trade a specified value +class OrderValueHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + order_value(data['stock'], float(data['value']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "order_value", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("order_value failed") + raise HTTPError(400, f"Order submission failed: {str(e)}") + +# order_percent() - Trade a specified percentage +class OrderPercentHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + order_percent(data['stock'], float(data['percent']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "order_percent", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("order_percent failed") + raise HTTPError(400, f"Order submission failed: {str(e)}") + +# order_target_value() - Trade to a target value +class OrderTargetValueHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + order_target_value(data['stock'], float(data['tar_value']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "order_target_value", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("order_target_value failed") + raise HTTPError(400, f"Order submission failed: {str(e)}") + +# order_target_percent() - Trade to a target percentage +class OrderTargetPercentHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + order_target_percent(data['stock'], float(data['tar_percent']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "order_target_percent", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("order_target_percent failed") + raise HTTPError(400, f"Order submission failed: {str(e)}") + +# order_shares() - Trade a specified number of shares +class OrderSharesHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + order_shares(data['stock'], int(data['shares']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "order_shares", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("order_shares failed") + raise HTTPError(400, f"Order submission failed: {str(e)}") + + +# ============= 5. Futures trading ============= +# buy_open() - Open a long futures position +class FuturesBuyOpenHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + buy_open(data['stock'], int(data['amount']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "buy_open", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("buy_open failed") + raise HTTPError(400, f"Failed to open a long futures position: {str(e)}") + +# buy_close_tdayfirst() - Close a short futures position, today first +class FuturesBuyCloseTdayFirstHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + buy_close_tdayfirst(data['stock'], int(data['amount']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "buy_close_tdayfirst", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("buy_close_tdayfirst failed") + raise HTTPError(400, f"Failed to close a short futures position (today first): {str(e)}") + +# buy_close_ydayfirst() - Close a short futures position, yesterday first +class FuturesBuyCloseYdayFirstHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + buy_close_ydayfirst(data['stock'], int(data['amount']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "buy_close_ydayfirst", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("buy_close_ydayfirst failed") + raise HTTPError(400, f"Failed to close a short futures position (yesterday first): {str(e)}") + +# sell_open() - Open a short futures position +class FuturesSellOpenHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + sell_open(data['stock'], int(data['amount']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "sell_open", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("sell_open failed") + raise HTTPError(400, f"Failed to open a short futures position: {str(e)}") + +# sell_close_tdayfirst() - Close a long futures position, today first +class FuturesSellCloseTdayFirstHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + sell_close_tdayfirst(data['stock'], int(data['amount']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "sell_close_tdayfirst", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("sell_close_tdayfirst failed") + raise HTTPError(400, f"Failed to close a long futures position (today first): {str(e)}") + +# sell_close_ydayfirst() - Close a long futures position, yesterday first +class FuturesSellCloseYdayFirstHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + sell_close_ydayfirst(data['stock'], int(data['amount']), data.get('style', 'LATEST'), + float(data.get('price', 0)), self.ctx(), data.get('accId', self.acc())) + self.write(json.dumps({"status": "success", "action": "sell_close_ydayfirst", "stock": data['stock']}, ensure_ascii=False)) + except Exception as e: + logger.exception("sell_close_ydayfirst failed") + raise HTTPError(400, f"Failed to close a long futures position (yesterday first): {str(e)}") + + +# ============= 6. Task management ============= +# cancel_task() - Cancel a task +class CancelTaskHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + taskId = data['taskId'] + accountType = data.get('accountType', 'stock') + ret = cancel_task(taskId, self.acc(), accountType, self.ctx()) + self.write(json.dumps({"status": "success" if ret else "failed", "taskId": taskId}, ensure_ascii=False)) + except Exception as e: + logger.exception("cancel_task failed") + raise HTTPError(400, f"Failed to cancel task: {str(e)}") + +# pause_task() - Pause a task +class PauseTaskHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + taskId = data['taskId'] + accountType = data.get('accountType', 'stock') + ret = pause_task(taskId, self.acc(), accountType, self.ctx()) + self.write(json.dumps({"status": "success" if ret else "failed", "taskId": taskId}, ensure_ascii=False)) + except Exception as e: + logger.exception("pause_task failed") + raise HTTPError(400, f"Failed to pause task: {str(e)}") + +# resume_task() - Resume a task +class ResumeTaskHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + taskId = data['taskId'] + accountType = data.get('accountType', 'stock') + ret = resume_task(taskId, self.acc(), accountType, self.ctx()) + self.write(json.dumps({"status": "success" if ret else "failed", "taskId": taskId}, ensure_ascii=False)) + except Exception as e: + logger.exception("resume_task failed") + raise HTTPError(400, f"Failed to resume task: {str(e)}") + +# do_order() - Trigger the previous bar's signal in real time +class DoOrderHandler(BaseHandler): + def post(self): + try: + do_order(self.ctx()) + self.write(json.dumps({"status": "success", "message": "Signal triggered"}, ensure_ascii=False)) + except Exception as e: + logger.exception("do_order failed") + raise HTTPError(400, f"Failed to trigger signal: {str(e)}") + + +# ============= 7. Account and order queries ============= +# Use fixed fields for high-frequency objects to avoid full dir/getattr reflection. +TRADE_DETAIL_FIELDS = ( + 'm_strOrderSysID', 'm_strInstrumentID', 'm_strExchangeID', + 'm_strInstrumentName', 'm_nOffsetFlag', 'm_nOrderStatus', + 'm_nVolumeTotal', 'm_nVolumeTraded', 'm_nOrderTime', + 'm_strInsertDate', 'm_strInsertTime', 'm_strRemark', + 'm_dPrice', 'm_dTradePrice', 'm_dTradeAmount', +) + + +def fixed_fields(obj, fields=TRADE_DETAIL_FIELDS): + result = {} + for field in fields: + try: + value = getattr(obj, field) + except (AttributeError, TypeError): + continue + if not callable(value): + result[field] = str(value) + if not result: + attrs = getattr(obj, '__dict__', {}) + result = { + key: str(value) for key, value in attrs.items() + if not key.startswith('_') and not callable(value) + } + return result + + +# get_trade_detail_data() - Get positions, orders, trades, or account details +class TradeDetailDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + account = data.get('account', 'stock') + datatype = data.get('datatype', 'position') + ret = safe_call(get_trade_detail_data, self.acc(), account, datatype) + if ret is None: + ret = [] + self.write(json.dumps({"data": ret}, ensure_ascii=False)) + +# get_value_by_order_id() - Get order or trade details by order ID +class ValueByOrderIdHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + orderId = data.get('orderId', '') + accountType = data.get('accountType', 'stock') + datatype = data.get('datatype', 'ORDER') + ret = safe_call(get_value_by_order_id, orderId, self.acc(), accountType, datatype) + attrs = fixed_fields(ret) if ret else {} + self.write(json.dumps({"orderId": orderId, "data": attrs}, ensure_ascii=False)) + +# get_last_order_id() - Get the latest order ID +class LastOrderIdHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + account = data.get('account', 'stock') + datatype = data.get('datatype', 'ORDER') + ret = safe_call(get_last_order_id, self.acc(), account, datatype) + self.write(json.dumps({"last_order_id": ret}, ensure_ascii=False)) + +# can_cancel_order() - Check whether an order can be canceled +class CanCancelOrderHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + orderId = data.get('orderId', '') + accountType = data.get('accountType', 'stock') + ret = safe_call(can_cancel_order, orderId, self.acc(), accountType) + self.write(json.dumps({"orderId": orderId, "can_cancel": ret}, ensure_ascii=False)) + + +class CancelByIdHandler(BaseHandler): + """Cancel an order by its actual system order ID.""" + def post(self): + data = json.loads(self.request.body) + order_id = str(data.get('order_id', '')).strip() + account_type = data.get('account_type', 'stock') + if not order_id: + raise HTTPError(400, "order_id cannot be empty") + cancelable = safe_call(can_cancel_order, order_id, self.acc(), account_type) + if not cancelable: + self.write(json.dumps({ + "status": "failed", "order_id": order_id, + "message": "Order does not exist or cannot currently be canceled" + }, ensure_ascii=False)) + return + result = safe_call(cancel, order_id, self.acc(), account_type, self.ctx()) + self.write(json.dumps({ + "status": "success" if result is not False else "failed", + "order_id": order_id, + }, ensure_ascii=False)) + +# get_debt_contract() - Get margin financing and securities lending debt contracts +class DebtContractHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + accId = data.get('accId', self.acc()) + ret = safe_call(get_debt_contract, accId) + result = [fixed_fields(obj) for obj in (ret or [])] + self.write(json.dumps({"data": result}, ensure_ascii=False)) + +# get_assure_contract() - Get margin collateral instruments +class AssureContractHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + accId = data.get('accId', self.acc()) + ret = safe_call(get_assure_contract, accId) + result = [fixed_fields(obj) for obj in (ret or [])] + self.write(json.dumps({"data": result}, ensure_ascii=False)) + +# get_enable_short_contract() - Get securities available for short selling +class EnableShortContractHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + accId = data.get('accId', self.acc()) + ret = safe_call(get_enable_short_contract, accId) + result = [fixed_fields(obj) for obj in (ret or [])] + self.write(json.dumps({"data": result}, ensure_ascii=False)) + +# get_ipo_data() - Get today's new stock and bond offerings +class IpoDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + typ = data.get('type', '') + ret = safe_call(get_ipo_data, typ) + self.write(json.dumps({"data": ret or {}}, ensure_ascii=False, default=str)) + +# get_new_purchase_limit() - Get the new-share subscription quota +class NewPurchaseLimitHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + accid = data.get('accid', self.acc()) + ret = safe_call(get_new_purchase_limit, accid) + self.write(json.dumps({"data": ret or {}}, ensure_ascii=False, default=str)) + + +# ============= 8. Reference functions (ext_data) ============= +# ext_data() - Get an extended-data value +class ExtDataHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + extdataname = data.get('extdataname', '') + stockcode = data.get('stockcode', '') + deviation = int(data.get('deviation', '0')) + ret = safe_call(ext_data, extdataname, stockcode, deviation, self.ctx()) + self.write(json.dumps({"extdataname": extdataname, "stockcode": stockcode, "value": ret}, ensure_ascii=False)) + +# ext_data_rank() - Get an extended-data ranking +class ExtDataRankHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + extdataname = data.get('extdataname', '') + stockcode = data.get('stockcode', '') + deviation = int(data.get('deviation', '0')) + ret = safe_call(ext_data_rank, extdataname, stockcode, deviation, self.ctx()) + self.write(json.dumps({"extdataname": extdataname, "stockcode": stockcode, "rank": ret}, ensure_ascii=False)) + +# get_factor_value() - Get a factor value +class GetFactorValueHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + factorname = data.get('factorname', '') + stockcode = data.get('stockcode', '') + deviation = int(data.get('deviation', '0')) + ret = safe_call(get_factor_value, factorname, stockcode, deviation, self.ctx()) + self.write(json.dumps({"factorname": factorname, "stockcode": stockcode, "value": ret}, ensure_ascii=False)) + +# get_factor_rank() - Get a factor ranking +class GetFactorRankHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + factorname = data.get('factorname', '') + stockcode = data.get('stockcode', '') + deviation = int(data.get('deviation', '0')) + ret = safe_call(get_factor_rank, factorname, stockcode, deviation, self.ctx()) + self.write(json.dumps({"factorname": factorname, "stockcode": stockcode, "rank": ret}, ensure_ascii=False)) + + +# ============= 9. Legacy handlers (compatibility) ============= +# get_trade_detail_data('position') - Query positions in the wrapped format +class HoldingHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + account = data.get('account', 'stock') + positions = safe_call(get_trade_detail_data, self.acc(), account, 'position') or [] + self.write(json.dumps({"data": positions}, ensure_ascii=False)) + +# get_trade_detail_data('account') - Query account assets +class AssetsHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + account = data.get('account', 'stock') + _data = safe_call(get_trade_detail_data, self.acc(), account, 'account') + info = _data[0] if _data else None + if not info: + raise HTTPError(500, "Failed to get account data") + self.write(json.dumps({"data": info}, ensure_ascii=False)) + + +# get_trade_detail_data('account') - Query total assets +class TotalMoneyHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + account = data.get('account', 'stock') + _data = safe_call(get_trade_detail_data, self.acc(), account, 'account') + info = _data[0] if _data else None + if not info: + raise HTTPError(500, "Failed to get account data") + self.write(json.dumps({"total_money": round(info.m_dBalance, 2)}, ensure_ascii=False)) + +# get_trade_detail_data('account') - Query available cash +class AvailableMoneyHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + account = data.get('account', 'stock') + _data = safe_call(get_trade_detail_data, self.acc(), account, 'account') + info = _data[0] if _data else None + if not info: + raise HTTPError(500, "Failed to get account data") + self.write(json.dumps({"available_money": round(info.m_dAvailable, 2)}, ensure_ascii=False)) + +# passorder(23) - Simplified buy order wrapper +class BuyHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + stock = data['stock'] + price = float(data['price']) + volume = int(data['volume']) + pr_type = data.get('prType', 11) + sn = data.get('sn','') + order_ref = passorder(23, 1101, self.acc(), stock, pr_type, price, volume, sn, 2, self.ctx()) + self.write(json.dumps({ + "status": "success", "action": "buy", "stock": stock, + "order_ref": str(order_ref) if order_ref else "unknown" + }, ensure_ascii=False)) + except Exception as e: + logger.exception("Buy order submission failed") + raise HTTPError(400, f"Order submission failed: {str(e)}") + +# passorder(24) - Simplified sell order wrapper +class SellHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + stock = data['stock'] + price = float(data['price']) + volume = int(data['volume']) + pr_type = data.get('prType', 11) + sn = data.get('sn','') + order_ref = passorder(24, 1101, self.acc(), stock, pr_type, price, volume, sn, 2, self.ctx()) + self.write(json.dumps({ + "status": "success", "action": "sell", "stock": stock, + "order_ref": str(order_ref) if order_ref else "unknown" + }, ensure_ascii=False)) + except Exception as e: + logger.exception("Sell order submission failed") + raise HTTPError(400, f"Order submission failed: {str(e)}") + +# get_trade_detail_data('order') - Query order statuses +class OrderStatusHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + account = data.get('account', 'stock') + sn = data.get('sn','') + orders = safe_call(get_trade_detail_data, self.acc(), account, 'order', sn) or [] + rets = [] + for order in orders: + rets.append({ + "order_sys_id": order.m_strOrderSysID, + "status": order.m_nOrderStatus, + "volume_left": order.m_nVolumeTotal, + "volume_traded": order.m_nVolumeTraded, + }) + self.write(json.dumps({"orders": rets}, ensure_ascii=False)) + +# cancel() - Cancel all orders +class CancelAllHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + account = data.get('account', 'stock') + sn = data.get('sn','') + orders = safe_call(get_trade_detail_data, self.acc(), account, 'order', sn) or [] + canceled_list = [] + for order in orders: + if can_cancel_order(order.m_strOrderSysID, self.acc(), account): + cancel(order.m_strOrderSysID, self.acc(), account, self.ctx()) + canceled_list.append({ + "order_sys_id": order.m_strOrderSysID, + "stock": order.m_strInstrumentID, + "volume_left": order.m_nVolumeTotal + }) + self.write(json.dumps({ + "status": "success", + "message": f"Submitted {len(canceled_list)} cancellation requests", + "canceled_orders": canceled_list + }, ensure_ascii=False)) + except Exception as e: + logger.exception("Cancel-all operation failed") + raise HTTPError(500, f"Order cancellation failed: {str(e)}") + + +class CancelByRuleHandler(BaseHandler): + def post(self): + try: + data = json.loads(self.request.body) + stock = data.get('stock') + sn = data.get('sn','') + cancel_volume = int(data.get('volume', 0)) + account = data.get('account', 'stock') + if not stock or cancel_volume <= 0: + raise HTTPError(400, "Invalid parameters: stock is required and volume must be greater than zero") + orders = safe_call(get_trade_detail_data, self.acc(), account, 'order', sn) or [] + target_orders = [] + for order in orders: + order_code = f"{order.m_strInstrumentID}.{order.m_strExchangeID}" + if order.m_nVolumeTotal + order.m_nVolumeTraded == cancel_volume and order_code == stock and can_cancel_order(order.m_strOrderSysID, self.acc(), account): + target_orders.append(order) + if not target_orders: + self.write(json.dumps({"status": "failed", "message": "No matching active orders found"}, ensure_ascii=False)) + return + canceled_ids = [] + for t_order in target_orders: + cancel(t_order.m_strOrderSysID, self.acc(), account, self.ctx()) + canceled_ids.append(t_order.m_strOrderSysID) + self.write(json.dumps({ + "status": "success", + "message": f"Matched {len(target_orders)} orders and submitted cancellation requests", + "canceled_sys_ids": canceled_ids + }, ensure_ascii=False)) + except Exception as e: + logger.exception("Rule-based cancellation failed") + raise HTTPError(500, f"Order cancellation failed: {str(e)}") + +# cancel() - Cancel orders by symbol and volume matching rules +# sys: Python version information +class PythonVersionHandler(BaseHandler): + def get(self): + import sys + version_info = { + "python_version": sys.version, + "python_version_info": { + "major": sys.version_info.major, + "minor": sys.version_info.minor, + "micro": sys.version_info.micro, + "releaselevel": sys.version_info.releaselevel, + "serial": sys.version_info.serial, + } + } + self.write(json.dumps(version_info, ensure_ascii=False)) + +# sys: Shut down the HTTP service +class ShutdownHandler(BaseHandler): + def post(self): + logger.info("Shutdown request received; the server is stopping...") + self.write(json.dumps({"status": "success", "message": "Server is shutting down..."}, ensure_ascii=False)) + self.finish() + IOLoop.current().add_callback(IOLoop.current().stop) + +# get_trade_detail_data('deal') - Query trade details +class DealHandler(BaseHandler): + def post(self): + data = json.loads(self.request.body) + account = data.get('account', 'stock') + sn = data.get('sn','') + deals = safe_call(get_trade_detail_data, self.acc(), account, 'deal', sn) or [] + rets = [fixed_fields(deal) for deal in deals] + self.write(json.dumps({"deals": rets}, ensure_ascii=False)) + + +# ============= Route registration ============= +def make_app(): + return Application([ + # V2 + (r"/api/v2/positions", HoldingHandler), + (r"/api/v2/assets", AssetsHandler), + + # Legacy compatibility routes + (r"/api/holding", HoldingHandler), + (r"/api/money/total", TotalMoneyHandler), + (r"/api/money/available", AvailableMoneyHandler), + (r"/api/order/buy", BuyHandler), + (r"/api/order/sell", SellHandler), + (r"/api/order/status", OrderStatusHandler), + (r"/api/order/cancel_all", CancelAllHandler), + (r"/api/order/cancel_order", CancelByRuleHandler), + (r"/api/order/cancel_by_id", CancelByIdHandler), + (r"/api/order/deal", DealHandler), + + # ContextInfo properties + (r"/api/context/period", ContextPeriodHandler), + (r"/api/context/barpos", ContextBarposHandler), + (r"/api/context/time_tick_size", ContextTimeTickSizeHandler), + (r"/api/context/stockcode", ContextStockCodeHandler), + (r"/api/context/dividend_type", ContextDividendTypeHandler), + (r"/api/context/market", ContextMarketHandler), + (r"/api/context/do_back_test", ContextDoBackTestHandler), + (r"/api/context/benchmark", ContextBenchmarkHandler), + (r"/api/context/capital", ContextCapitalHandler), + (r"/api/context/universe", ContextUniverseHandler), + + # Data queries + (r"/api/data/stock_name", StockNameHandler), + (r"/api/data/open_date", OpenDateHandler), + (r"/api/data/last_volume", LastVolumeHandler), + (r"/api/data/bar_timetag", BarTimetagHandler), + (r"/api/data/tick_timetag", TickTimetagHandler), + (r"/api/data/sector", SectorHandler), + (r"/api/data/industry", IndustryHandler), + (r"/api/data/stock_list_in_sector", StockListInSectorHandler), + (r"/api/data/weight_in_index", WeightInIndexHandler), + (r"/api/data/contract_multiplier", ContractMultiplierHandler), + (r"/api/data/risk_free_rate", RiskFreeRateHandler), + (r"/api/data/date_location", DateLocationHandler), + (r"/api/data/history_data", HistoryDataHandler), + (r"/api/data/market_data", MarketDataHandler), + (r"/api/data/market_data_ex", MarketDataExHandler), + (r"/api/data/full_tick", FullTickHandler), + (r"/api/data/divid_factors", DividFactorsHandler), + (r"/api/data/main_contract", MainContractHandler), + (r"/api/data/timetag_to_datetime", TimetagToDatetimeHandler), + (r"/api/data/total_share", TotalShareHandler), + (r"/api/data/trading_dates", TradingDatesHandler), + (r"/api/data/svol", SvolHandler), + (r"/api/data/bvol", BvolHandler), + (r"/api/data/longhubang", LonghubangHandler), + (r"/api/data/top10_share_holder", Top10ShareHolderHandler), + (r"/api/data/option_detail", OptionDetailHandler), + (r"/api/data/turnover_rate", TurnoverRateHandler), + (r"/api/data/etf_info", EtfInfoHandler), + (r"/api/data/etf_iopv", EtfIopvHandler), + (r"/api/data/instrumentdetail", InstrumentDetailHandler), + (r"/api/data/contract_expire_date", ContractExpireDateHandler), + (r"/api/data/option_undl_data", OptionUndlDataHandler), + (r"/api/data/financial_data", FinancialDataHandler), + (r"/api/data/factor_data", FactorDataHandler), + (r"/api/data/his_st_data", HisStDataHandler), + (r"/api/data/his_index_data", HisIndexDataHandler), + (r"/api/data/all_subscription", AllSubscriptionHandler), + (r"/api/data/option_list", OptionListHandler), + (r"/api/data/his_contract_list", HisContractListHandler), + (r"/api/data/option_iv", OptionIvHandler), + (r"/api/data/bsm_price", BsmPriceHandler), + (r"/api/data/bsm_iv", BsmIvHandler), + (r"/api/data/local_data", LocalDataHandler), + + # Subscriptions + (r"/api/data/subscribe_quote", SubscribeQuoteHandler), + (r"/api/data/unsubscribe_quote", UnsubscribeQuoteHandler), + + # Predicate functions + (r"/api/check/is_last_bar", IsLastBarHandler), + (r"/api/check/is_new_bar", IsNewBarHandler), + (r"/api/check/is_suspended_stock", IsSuspendedStockHandler), + (r"/api/check/is_sector_stock", IsSectorStockHandler), + (r"/api/check/is_typed_stock", IsTypedStockHandler), + (r"/api/check/get_industry_name_of_stock", GetIndustryNameOfStockHandler), + + # Trading + (r"/api/trade/passorder", PassorderHandler), + (r"/api/trade/algo_passorder", AlgoPassorderHandler), + (r"/api/trade/smart_algo_passorder", SmartAlgoPassorderHandler), + (r"/api/trade/order_lots", OrderLotsHandler), + (r"/api/trade/order_value", OrderValueHandler), + (r"/api/trade/order_percent", OrderPercentHandler), + (r"/api/trade/order_target_value", OrderTargetValueHandler), + (r"/api/trade/order_target_percent", OrderTargetPercentHandler), + (r"/api/trade/order_shares", OrderSharesHandler), + + # Futures trading + (r"/api/trade/futures/buy_open", FuturesBuyOpenHandler), + (r"/api/trade/futures/buy_close_tdayfirst", FuturesBuyCloseTdayFirstHandler), + (r"/api/trade/futures/buy_close_ydayfirst", FuturesBuyCloseYdayFirstHandler), + (r"/api/trade/futures/sell_open", FuturesSellOpenHandler), + (r"/api/trade/futures/sell_close_tdayfirst", FuturesSellCloseTdayFirstHandler), + (r"/api/trade/futures/sell_close_ydayfirst", FuturesSellCloseYdayFirstHandler), + + # Task management + (r"/api/trade/cancel_task", CancelTaskHandler), + (r"/api/trade/pause_task", PauseTaskHandler), + (r"/api/trade/resume_task", ResumeTaskHandler), + (r"/api/trade/do_order", DoOrderHandler), + + # Account and order queries + (r"/api/trade/trade_detail_data", TradeDetailDataHandler), + (r"/api/trade/value_by_order_id", ValueByOrderIdHandler), + (r"/api/trade/last_order_id", LastOrderIdHandler), + (r"/api/trade/can_cancel_order", CanCancelOrderHandler), + (r"/api/trade/debt_contract", DebtContractHandler), + (r"/api/trade/assure_contract", AssureContractHandler), + (r"/api/trade/enable_short_contract", EnableShortContractHandler), + (r"/api/trade/ipo_data", IpoDataHandler), + (r"/api/trade/new_purchase_limit", NewPurchaseLimitHandler), + + # Reference functions + (r"/api/ext/ext_data", ExtDataHandler), + (r"/api/ext/ext_data_rank", ExtDataRankHandler), + (r"/api/ext/get_factor_value", GetFactorValueHandler), + (r"/api/ext/get_factor_rank", GetFactorRankHandler), + + # System + (r"/api/sys/python_version", PythonVersionHandler), + (r"/api/sys/shutdown", ShutdownHandler), + + ], debug=False) + +# ============= Callback registration ============= +def json_serializer(obj): + if isinstance(obj, datetime.datetime): + return obj.strftime("%Y-%m-%d %H:%M:%S") + if isinstance(obj, datetime.date): + return obj.strftime("%Y-%m-%d") + if hasattr(obj, 'to_dict'): + return obj.to_dict() + attrs = getattr(obj, '__dict__', None) + if attrs: + return {key: value for key, value in attrs.items() if not key.startswith('_')} + return str(obj) + +def write_json(file_key, data,order_id:str=''): + """ + Write data to a JSON file. + + Args: + file_key (str): File-name template. + data: Data to write. + order_id (str): Optional order ID. + """ + now=datetime.datetime.now() + day = now.strftime("%Y%m%d") + body = { + "created_at": now, + "body":data + } + # Build the target path. + if order_id: + file_path = Path(DATA_DIR) / (file_key % (day,order_id)) + else: + file_path = Path(DATA_DIR) / (file_key % day) + + # Create the target directory. + file_path.parent.mkdir(parents=True, exist_ok=True) + + temporary_path = file_path.with_suffix(file_path.suffix + '.tmp') + with open(temporary_path, 'w', encoding='utf-8') as f: + json.dump(body, f, ensure_ascii=False, separators=(',', ':'), default=json_serializer) + f.flush() + os.fsync(f.fileno()) + os.replace(temporary_path, file_path) + +# Account update callback +def account_callback(ContextInfo, accountInfo): + write_json("acount_%s.json",accountInfo) + +# Order update callback +def order_callback(ContextInfo, orderInfo): + write_json("order_%s_%s.json",orderInfo,orderInfo.m_strOrderSysID) + +# Trade update callback +def deal_callback(ContextInfo, dealInfo): + write_json("deal_%s_%s.json",dealInfo,dealInfo.m_strOrderSysID) + +# Position update callback +def position_callback(ContextInfo, positonInfo): + write_json("position_%s.json",positonInfo) + +# Order error callback +def orderError_callback(ContextInfo, passOrderInfo, msg): + print('orderError_callback') + # Log order and error details. + print (passOrderInfo.orderCode) + print (msg) + +def init(ContextInfo): + if not (ACCOUNT_ID or "").strip(): + msg = "ACCOUNT_ID is empty; startup aborted" + logger.error(msg) + raise ValueError(msg) + if not (DATA_DIR or "").strip(): + msg = "DATA_DIR is empty; startup aborted" + logger.error(msg) + raise ValueError(msg) + try: + ContextInfo.accountID = ACCOUNT_ID + ContextInfo.set_account(ACCOUNT_ID) + # Load the symbol universe only when configured. + pass_codes_path = Path(DATA_DIR) / "pass_codes.json" + with pass_codes_path.open("r", encoding="utf-8") as stream: + codes = json.load(stream) + ContextInfo.set_universe(list(codes)) + + # Api App + app = make_app() + app.ContextInfo = ContextInfo + app.accountID = ContextInfo.accountID + app.listen(PORT, address='0.0.0.0') + logger.info(f"ACCOUNT_ID: {ACCOUNT_ID}") + logger.info(f"DATA_DIR: {DATA_DIR}") + logger.info(f"TOKEN: {TOKEN}") + logger.info(f"Initialized symbol universe with {len(codes)} instruments") + logger.info(f"QMT HTTP Server started at http://0.0.0.0:{PORT} (all APIs loaded)") + IOLoop.current().start() + except Exception as e: + logger.exception(f"server start failed: {e}") diff --git a/api/QMT_API.py b/api/qmt_api_rele.py similarity index 100% rename from api/QMT_API.py rename to api/qmt_api_rele.py diff --git a/docs/CODE_AUDIT_2026-08-29.md b/docs/CODE_AUDIT_2026-08-29.md new file mode 100644 index 0000000..56cf82c --- /dev/null +++ b/docs/CODE_AUDIT_2026-08-29.md @@ -0,0 +1,204 @@ +# big-qmt 代码复审报告（2026-08-29） + +## 1. 审计范围与方法 + +- 审计范围：当前工作区中的 `api/`、`py-client/`、启动脚本及测试。 +- 明确排除：`docs/todo.md`，本报告没有读取或引用该文件作为判断依据。 +- 关注项：致命错误、交易业务逻辑错误、代码冗余、过度验证。 +- 动态验证：`python -B -m unittest discover -s tests -v`；共 11 项，8 项通过、3 项错误。 +- 编译验证：`python -B -m compileall -q .` 通过。 +- 限制：未连接真实 QMT，未执行真实委托、撤单或成交回报验证。 + +## 2. 总结 + +当前版本不适合直接实盘运行。至少存在 4 条会使计划业务完全不执行或造成重复/超额下单的高风险路径： + +1. IPO 功能在启用时必然先抛出 `TypeError`，而且其后仍有路径拼接、锁判断和调度生命周期错误。 +2. 趋势策略的 `UNKNOWN` 状态虽然不会被删除，但并未参与开仓过滤，无法兑现“禁止自动重复下单”。 +3. 多个开仓信号之间没有共享可用资金预算，且首笔开仓数量函数可能主动超过配置金额。 +4. `OrderBook` 会覆盖同证券同方向的多笔订单，并可能反复撤销已终结的历史订单。 + +测试失败并非测试环境问题，而是直接命中了当前生产函数的确定性参数错误。 + +## 3. P0：致命或可能造成错误交易 + +### 3.1 IPO 入口启用后必然报错，全部核心测试无法执行 + +位置： + +- `py-client/strategy/ipo/boot.py:22-29` +- `py-client/libs/calc.py:5-7` + +`AutoBuyIpo(now)` 调用 `trading_time()` 时没有传入必需的 `now` 参数。只要 `enable_auto_ipo=True`，函数便在创建客户端、读取候选和检查重复委托之前抛出 `TypeError`。 + +复现结果： + +- `test_local_record_prevents_duplicate_after_restart`：ERROR +- `test_broker_order_prevents_duplicate`：ERROR +- `test_one_rejection_does_not_stop_other_candidates`：ERROR + +影响：自动打新完全不可用；如果由主程序同步调用，异常还可能中止启动流程。 + +建议：传入 `now or datetime.now()`，并将交易日与交易时段判断统一成一个可测试入口。修复后必须重新运行现有 4 项 IPO 测试。 + +### 3.2 IPO 路径还有两层确定性错误：字符串路径相除、锁语义反向 + +位置： + +- `py-client/strategy/ipo/boot.py:37-49` +- `py-client/config/__init__.py:28,109` +- `py-client/libs/lockfile.py:9-18` + +即使修复 3.1，代码仍执行： + +```python +Path(config.global_config.qmt_data_dir / f"{stock}.lock") +``` + +`qmt_data_dir` 在 `GlobalConfig` 中是 `str`，因此先计算 `str / str`，会再次抛出 `TypeError`。此外，`is_lock()` 的语义是“锁文件已存在”，当前代码却仅在其返回 `True` 时提交申购；首次运行没有锁文件，所有候选都会被跳过。正确业务语义应是“没有本地成功记录时才继续”，并且还要和券商订单/成交记录交叉核对。 + +影响：修复第一个异常后，IPO 仍无法正常首申购；若人工预建锁文件绕过判断，则反而允许对已锁定证券再次申购。 + +建议：使用 `Path(qmt_data_dir) / f"{stock}.lock"`；反转本地锁判断；提交成功后才写锁；提交失败不得写锁；恢复测试中已经表达但当前实现缺失的券商订单和成交去重。 + +### 3.3 `UNKNOWN` 只被保存在状态文件中，没有真正阻止再次开仓 + +位置： + +- `py-client/strategy/trend/state.py:76-82,175-228` +- `py-client/strategy/trend/boot.py:136-164` +- `py-client/strategy/trend/open.py:17-70` + +`State.delete()` 会静默保留 `ING/UNKNOWN`，但 `RunOnce()` 构造 `allow_open` 时只排除真实持仓代码，不排除 `State` 中的未决代码。`open_signal()` 也只检查 `OrderBook.busy()`。当重启对账无法找到真实活动委托并把订单转成 `UNKNOWN` 后，如果券商订单查询暂时缺失或委托已终结但结果不明，`OrderBook` 没有活动锁，下一次同证券信号仍可提交新买单。 + +另外，`RunOnce()` 无论 `State.delete()` 是否因保护而拒绝删除，都会执行 `open_watch.forget()` 和 `add_watch.forget()`，形成“状态保留但观察锁被清除”的不一致状态。 + +影响：前一笔订单结果无法确认时仍可能重复开仓；这与状态修复目标和验收标准直接冲突。 + +建议：开仓候选必须同时排除持仓代码、`ING/UNKNOWN` 状态代码以及活动买单代码。`delete()` 应返回是否实际删除，调用方只在删除成功后清理观察器。运行期每轮应将订单簿刷新结果用于状态对账，而不是仅在启动时对账一次。 + + +## 4. P1：重要业务逻辑错误 + +### 4.1 IPO 定时任务实际上不会在 10:00 被调度 + +位置：`py-client/main.py:111-115` + +代码注册每日 10:00 任务后只调用一次 `schedule.run_pending()`，随后进入 `StartTrend()` 的永久循环，再也没有机会驱动调度器。除非程序恰好在任务已到期的极窄窗口启动，否则自动打新不会执行。 + +建议：把调度轮询合并进主循环，或使用独立受控线程/独立进程；IPO 异常必须隔离，不能终止趋势策略。 + +### 4.2 样例账户没有启用任何趋势信号 + +位置： + +- `py-client/config/__init__.py:48` +- `py-client/etc/dev.yaml` +- `py-client/libs/signal.py:23-28` + +`signal_allow` 默认空列表，而当前 `dev.yaml` 没有该字段。`init_signals()` 只加载显式出现在允许列表中的信号，因此样例配置启动后趋势策略永远没有开仓候选，日志也不会提示“允许列表为空”。 + +建议：明确选择一种语义：空列表表示全部禁用，并在启动时显著告警；或空列表表示允许全部。样例配置应与预期业务一致。 + +### 4.3 同证券同方向的多笔委托会互相覆盖 + +位置：`py-client/strategy/trend/order.py:68-77,118-122` + +订单缓存以 `BUY-code` / `SELL-code` 为唯一键。券商返回同一证券同方向的多笔委托时，字典推导只保留最后一笔。过期撤单、状态展示和锁判断无法看到被覆盖的订单。 + +建议：以系统委托号作为主键，另建 `(side, code) -> set[system_order_id]` 活动索引。 + +### 4.4 过期撤单没有过滤活动状态 + +位置：`py-client/strategy/trend/order.py:79-90` + +`cancel_expired()` 遍历 `data.values()` 时只检查时间和系统委托号，没有要求 `order.status in BUSY_STATUSES`。如果接口返回历史委托，程序每轮都可能对已成交、已撤销或已失败订单调用撤单。 + +建议：只对活动状态且超过时限的订单撤单，并维护撤单请求中的冷却状态，避免每 30 秒重复请求。 + +### 4.5 状态落盘失败后的故障策略不一致 + +位置： + +- `py-client/strategy/trend/open.py:59-70` +- `py-client/strategy/trend/positions.py:175-183` + +开仓提交成功但保存失败时只记录日志，仍清除观察器并继续运行；补仓保存失败则异常向外传播，但真实订单已经提交。两条路径都可能出现“真实订单存在、本地恢复信息缺失”，处理策略却不一致。 + +建议：真实下单成功后若持久化失败，应将证券置于内存级熔断集合，保留订单簿锁并持续重试落盘；禁止该证券继续自动交易，直至对账确认。 + +### 4.6 启动脚本会杀死机器上的所有 Python 进程 + +位置：`run.bat:1-7` + +`taskkill /IM python.exe /F` 不限定当前项目、PID 或命令行，会强制终止机器上所有 Python 工作负载。脚本也没有先切换到自身目录，`python main.py` 是否能找到文件取决于调用时当前目录；仓库根目录本身并不存在 `main.py`。 + +影响：可能中断无关服务、研究任务或其他交易程序，同时自身仍可能因工作目录错误无法启动。 + +建议：保存本项目 PID 并只终止该 PID；脚本开头使用 `cd /d "%~dp0py-client"` 或调用绝对脚本路径；不要在生产启动流程中无条件 `git pull`。 + +## 5. P2：代码冗余、失效代码和过度验证 + +### 5.1 API 服务文件完全重复且正式入口不明确 + +位置： + +- `api/qmt_api_new.py` +- `api/qmt_api_rele.py` +- 工作区状态中的已删除 `api/QMT_API.py` + +两个现存文件 SHA-256 完全一致，属于逐字节重复；原正式命名文件当前又处于删除状态。部署人员无法从启动脚本或文档可靠判断应加载哪个版本，后续修复也容易只改到其中一份。 + +建议：保留唯一正式入口，开发版本通过 Git 分支管理，不以 `_new`、`_rele` 复制整文件。 + +### 5.2 IPO 模块呈现“旧实现覆盖新设计”的死代码特征 + +位置：`py-client/strategy/ipo/boot.py:5-19` + +`json`、`Any`、`IPO_STRATEGY_NAME`、`IPO_REMARKS`、`IPO_SESSIONS`、`TRADING_CALENDAR_SYMBOL` 均未使用；测试则期待交易日查询、券商对账、异常隔离、上下文管理器和返回成功数量，但当前函数都未实现。这不是单纯格式问题，而是实现与测试/设计发生大段脱节的信号。 + +建议：不要逐个删除常量掩盖问题；先恢复完整 IPO 流程，再移除确认无用的符号。 + +### 5.3 主程序存在不可达代码、拼写错误和未使用符号 + +位置： + +- `py-client/main.py:5,15,100-101,115-116` + +`StartTrend()` 正常情况下永久循环，因此其后的“策略启动成功”日志不可达；即便未来返回，使用的 `config.account_config.strateg` 也不存在。非 Windows 分支调用未定义的 `log.error`。`TimedRotatingFileHandler` 和 `GLOBAL_CONFIG_PATH` 没有使用。 + +建议：启动成功日志应放在进入循环前；统一使用 `logging`；清除未使用导入和常量。 + +### 5.4 `State.delete()` 的全局保护属于过度且不透明的验证 + +位置：`py-client/strategy/trend/state.py:76-82` + +任何调用者请求删除 `ING/UNKNOWN` 状态都会被静默拒绝，没有返回值、日志、强制删除入口或订单证据参数。它确实避免了一类误删，但也会阻止人工确认后的清理，并使调用者误以为删除成功。当前 `RunOnce()` 正因此错误清除了观察器。 + +建议：将“是否可清理”的业务判断放在显式对账流程中；`delete()` 返回布尔值或抛出明确异常；如需保护，提供带审计原因的显式强制路径。验证应基于订单证据，不应仅基于状态字符串。 + +### 5.5 对账存在重复落盘 + +位置：`py-client/strategy/trend/state.py:84-108,110-137` + +`reconcile()` 先调用会自行 `save()` 的 `sync_positions()`，完成订单对账和清理后又 `save()`。每次启动对账至少写两次同一状态文件，第一份还是未完成订单对账的中间状态。 + +建议：为 `sync_positions()` 增加不立即保存的内部版本，整个对账事务只在最终一致状态落盘一次。 + +## 6. 建议整改顺序 + +1. 先恢复 IPO 的可执行性，并让现有 4 项 IPO 测试全部通过。 +2. 完成 `UNKNOWN/ING` 与开仓候选、订单簿之间的统一锁定，增加重启未成交端到端测试。 +3. 修复开仓数量和同轮资金预留，增加高价股、多信号资金边界测试。 +4. 重构 `OrderBook` 主键与活动索引，并限制撤单状态。 +5. 修复调度生命周期和启动脚本的进程范围。 +6. 最后清理重复 API 文件、死代码、重复落盘和无效日志。 + +## 7. 验收门槛 + +- 全部 11 项现有测试通过，且不通过跳过/删除失败测试达成。 +- 新增：IPO 首次申购、重复启动、券商已有委托、单只拒单不影响其他候选。 +- 新增：开仓未成交重启、订单查询暂时缺失、部分成交活动、部分成交撤单、完全成交。 +- 新增：`buy_value` 不足一手、多信号总金额超过现金、价格滑点场景。 +- 新增：同证券同方向两笔活动委托都能被发现和撤销。 +- 在模拟账户完成至少一次完整的下单、部分成交、撤单、重启对账闭环后，再考虑实盘。 diff --git a/docs/CODE_REAUDIT_2026-08-29_V2.md b/docs/CODE_REAUDIT_2026-08-29_V2.md new file mode 100644 index 0000000..c729b62 --- /dev/null +++ b/docs/CODE_REAUDIT_2026-08-29_V2.md @@ -0,0 +1,176 @@ +# big-qmt 手动修改后代码复审（2026-08-29 V2） + +## 1. 范围与验证 + +- 基于当前未提交工作区重新审计，不直接继承上一版报告结论。 +- 审计范围：`api/`、`py-client/`、启动脚本、配置和测试。 +- 明确排除：`docs/todo.md`；未读取、未引用其内容。 +- `python -B -m unittest discover -s tests -v`：11 项中 7 项通过、4 项错误。 +- 全部 Python 文件 AST 解析通过。 +- 未连接真实 QMT，未执行真实申购、买卖、撤单和成交回报测试。 + +## 2. 总体结论 + +当前版本仍不建议直接实盘运行。 + +手动修改已经修复了 IPO 路径中的三个表面问题：`trading_time()` 现在传入时间、数据目录能够正确转成 `Path`、首次申购的本地锁判断方向已经改正；APScheduler 也能在趋势策略永久循环之外每日 10:00 触发任务。 + +但核心安全闭环仍未完成：IPO 会在没有确认券商受理的情况下写入“已申购”锁，也没有使用券商委托/成交记录防重；趋势策略的 `UNKNOWN` 状态仍不能阻止再次开仓；开仓资金仍可能超预算。测试失败数量还从上一轮的 3 项变成了 4 项。 + +## 3. P0：致命或可能导致错误交易 + +### 3.1 IPO 不校验下单结果就写入成功锁，可能永久漏申购 + +位置：`py-client/strategy/ipo/boot.py:37-51` + +`client.passorder()` 的返回值被完全忽略，随后无条件执行 `write_lockfile(lp)`。如果服务端以正常 HTTP 响应返回 `{"status": "failed"}`、空订单号或其他业务拒绝结果，本地仍会记录为已申购，后续每天都会跳过该证券。 + +影响：券商实际没有接受订单，但本地永久认为已经提交，造成漏申购。 + +建议：统一校验 `status == "success"` 且 `order_ref` 有效；只有明确受理后才能写锁。未知响应应告警并保持可对账状态，不能直接标记成功。 + + + +### 3.3 单只 IPO 异常会中止当天全部后续候选 + +位置：`py-client/strategy/ipo/boot.py:37-51` + +候选循环内部没有单只证券级异常隔离。第一只证券的字段缺失、下单拒绝抛异常或锁文件写入失败，都会直接退出 `AutoBuyIpo()`；后面的候选不会再尝试。APScheduler 会记录任务异常，但当天 10:00 不会自动重新执行整个任务。 + +影响：一只异常证券导致当天其他所有新股漏申购。 + +建议：每只候选独立 `try/except` 并记录证券代码；失败继续处理下一只。任务结束后汇总成功、跳过、失败数量。 + +### 3.4 趋势策略 `UNKNOWN` 状态仍不能阻止自动重复开仓 + +位置： + +- `py-client/strategy/trend/state.py:83-89,184-237` +- `py-client/strategy/trend/boot.py:145-164` +- `py-client/strategy/trend/open.py:17-70` + +`State.delete()` 会保留 `ING/UNKNOWN`，但 `RunOnce()` 生成开仓候选时只排除真实持仓；`open_signal()` 只检查 `OrderBook.busy()`。如果启动对账将订单标记成 `UNKNOWN`，同时券商活动委托列表暂时没有该订单，状态文件虽然存在，下一轮信号仍可再次下单。 + +此外，运行期调用 `delete()` 后不检查是否真的删除，就清除两个观察器，造成未决状态与观察状态不一致。 + +影响：订单结果无法确认时可能重复买入，未达到“UNKNOWN 禁止自动重复下单”的目标。 + +建议：开仓候选同时排除 `ING/UNKNOWN` 状态代码和活动买单代码；`delete()` 返回实际删除结果；每轮用最新订单簿重新对账状态。 + + + +## 4. P1：重要业务逻辑问题 + +### 4.1 IPO 函数契约退化，现有 4 项测试全部报错 + +位置： + +- `py-client/strategy/ipo/boot.py:22` +- `py-client/tests/test_ipo.py:52-136` + +`AutoBuyIpo(now: datetime | None = None)` 被改成无参数函数，测试无法注入确定时间，4 项测试全部以 `TypeError: AutoBuyIpo() takes 0 positional arguments but 1 was given` 结束。APScheduler 并不要求删除可选参数；保留可选 `now` 同样可以无参调度。 + +测试当前也明确要求但实现未满足：返回成功数量、关闭客户端、本地重启防重、券商记录防重、单只拒绝不影响其他候选。 + +建议：恢复可选 `now` 参数，内部使用 `current = now or datetime.now()`；不要修改测试来掩盖业务契约缺失。 + + +### 4.3 IPO 客户端从不关闭 + +位置：`py-client/strategy/ipo/boot.py:31-51` + +每日任务创建新的 `httpx.Client`，成功、跳过和异常路径都没有调用 `close()`。长期运行会累积未及时释放的连接池资源。 + +建议：使用 `with Client(...) as client:`，现有 `Client` 已实现上下文管理器。 + +### 4.4 IPO 返回类型与真实返回值不一致 + +位置：`py-client/strategy/ipo/boot.py:22-51` + +函数标注返回 `int`，只有禁用和非交易时间返回 0；正常处理完候选后隐式返回 `None`，也没有统计提交成功数量。 + +影响：日志、监控和测试无法知道任务到底提交了多少只证券。 + +建议：维护 `success_count` 并在所有出口返回整数。 + +### 4.5 同证券同方向的多笔趋势委托仍会互相覆盖 + +位置：`py-client/strategy/trend/order.py:70-79,121-125` + +订单缓存仍以 `BUY-code` / `SELL-code` 为唯一键。同证券同方向多笔委托只保留最后一笔，其他活动订单无法撤销或对账。 + +建议：系统委托号作为主键，方向与证券组合作为一对多活动索引。 + +### 4.6 过期撤单仍会处理已终结历史订单 + +位置：`py-client/strategy/trend/order.py:81-93` + +撤单条件没有检查 `order.status in BUSY_STATUSES`。接口若返回历史订单，程序会对已成交、已撤、已失败订单反复发送撤单请求。 + +建议：仅处理活动状态，并为已发送撤单请求增加冷却或本地状态。 + +### 4.7 下单成功但状态落盘失败的安全策略不统一 + +位置： + +- `py-client/strategy/trend/open.py:59-70` +- `py-client/strategy/trend/positions.py:175-183` + +开仓保存失败只记日志并继续；补仓保存失败则异常退出本轮。两者都可能已经存在真实订单，却缺少可恢复的本地状态。 + +建议：落盘失败后将证券加入内存熔断集合、保留订单锁并持续重试；在完成真实订单对账前禁止该证券继续自动交易。 + + +### 5.3 主程序仍有不可达代码、拼写错误和无效符号 + +位置： + +- `py-client/main.py:5,15,100-101,127-128` + +`StartTrend()` 正常情况下永久循环，其后的日志不可达；即便返回，`config.account_config.strateg` 也不存在。非 Windows 分支使用未定义的 `log.error`。`TimedRotatingFileHandler`、`GLOBAL_CONFIG_PATH` 未使用。 + +建议：启动成功日志放在进入永久循环前；统一使用 `logging`；删除无效导入和常量。 + +### 5.4 `State.delete()` 的保护过宽且静默 + +位置：`py-client/strategy/trend/state.py:83-89` + +任何 `ING/UNKNOWN` 都会让删除静默失效，没有返回值、日志、订单证据或人工强制清理入口。这是基于状态字符串的全局拦截，不是真实订单对账，并已造成调用方误清观察器。 + +建议：把清理许可放入显式对账决策；`delete()` 返回布尔值或明确拒绝原因；人工确认终结后应有可审计的清理路径。 + +### 5.5 状态对账重复写盘并暴露中间状态 + +位置：`py-client/strategy/trend/state.py:91-117,119-146` + +`reconcile()` 调用会自行保存的 `sync_positions()`，完成订单对账后再次保存。一次启动对账至少写盘两次，第一次还是未完成订单状态恢复的中间结果。 + +建议：内部同步只改内存，完整对账结束后一次性原子落盘。 + +### 5.6 调度器配置包含当前进程内不必要的重复任务替换 + +位置：`py-client/main.py:116-123` + +调度器每次启动都是新实例，只添加一次固定 ID 任务，因此 `replace_existing=True` 在当前结构下没有实际作用。它无害，但属于多余防御参数，容易让人误以为使用了持久化任务仓库或存在重复注册路径。 + +建议：若没有持久化 job store 或重复注册，删除该参数；若未来启用持久化，再保留并补充任务版本策略。 + +## 6. 本轮已确认改善 + +- APScheduler 后台调度不会被 `StartTrend()` 的永久循环阻塞。 +- `timezone="Asia/Shanghai"` 明确了每日 10:00 的业务时区。 +- `coalesce=True` 和 `max_instances=1` 能防止任务积压补跑和并发重叠。 +- `dev.yaml` 已显式配置 `signal_allow`，趋势信号不再因默认空列表而全部禁用。 +- IPO 本地路径拼接和首次锁判断方向已经修正。 +- 趋势部分成交判定的四类状态逻辑仍保留。 + +## 7. 建议整改顺序与验收门槛 + +1. 恢复 IPO 可选时间参数、成功计数和上下文管理器，让现有 4 项 IPO 测试先全部通过。 +2. 增加真实交易日、券商委托/成交防重、下单结果校验和单只异常隔离。 +3. 让 `ING/UNKNOWN` 真正参与趋势开仓过滤，并补充重启未成交端到端测试。 +4. 修复单笔与同轮开仓资金预算。 +5. 重构订单簿一对多索引并限制撤单状态。 +6. 清理启动脚本、重复 API 文件和死代码。 + +最低验收要求：现有 11 项测试全部通过；新增 IPO 返回失败但不写锁、锁丢失但券商已有订单、第一只拒绝而第二只成功、节假日不申购；新增趋势 `UNKNOWN` 无持仓且无活动委托时仍不重复开仓；最后在模拟账户完成下单、部分成交、撤单、重启对账闭环。 diff --git a/docs/CODE_REAUDIT_2026-08-29_V3.md b/docs/CODE_REAUDIT_2026-08-29_V3.md new file mode 100644 index 0000000..1ccc7d2 --- /dev/null +++ b/docs/CODE_REAUDIT_2026-08-29_V3.md @@ -0,0 +1,249 @@ +# big-qmt 手动修改后代码复审（2026-08-29 V3） + +## 1. 审计范围与验证 + +- 基于当前未提交工作区重新审计，不直接复制上一版报告。 +- 范围：`api/`、`py-client/`、配置、启动脚本和测试。 +- 明确排除：`docs/todo.md`；本轮未读取、未引用其内容。 +- 全套测试：`python -B -m unittest discover -s tests -v`。 +- 测试结果：12 项中 8 项通过、4 项错误；错误全部位于 IPO 测试。 +- 全部 Python 文件 AST 解析通过。 +- 未连接真实 QMT，未执行真实委托、成交和撤单。 + +## 2. 总体结论 + +当前版本仍不建议直接实盘运行。 + +本轮有两项明确改善：趋势策略已在每轮重新对账，`ING/UNKNOWN` 同时参与候选过滤和下单前检查；订单方向锁也已经改成带秒级时间戳的字典并自动清理。新增的 `UNKNOWN` 无持仓防重复测试通过。 + +但订单锁的新过期规则会在真实委托仍活动时强制解锁，券商刷新还会整体覆盖本地下单锁；卖出订单没有持久化状态兜底，因此存在重复卖出风险。IPO 仍然存在“业务失败响应也写成功锁”、券商侧防重缺失和函数契约错误。资金预算、订单覆盖及启动脚本风险也尚未解决。 + +## 3. P0：致命或可能造成错误交易 + +### 3.1 活动委托达到 180 秒会被强制解锁，可能重复下单 + +位置：`py-client/strategy/trend/order.py:47-52,59-89,138-146` + +业务锁是否有效只由创建时间和 `lock_timeout_sec` 决定，不再参考订单是否仍属于 `BUSY_STATUSES`。`refresh()` 明明从券商读到活动订单，仍会把创建时间超过 180 秒的锁立即删除。 + +定向复现：状态为 `50` 的活动卖单，创建 181 秒后，`busy("A", "SELL")` 返回 `False`。 + +开仓和补仓还有 `State` 的 `ING/UNKNOWN` 兜底，但止盈卖单没有写入持久化状态。活动卖单超时或撤单失败后，下一轮止盈判断可能再次提交同一证券卖单。 + +建议：本地“提交防抖锁”可以超时，但券商明确返回活动状态时必须继续视为 busy；将两者拆成 `local_locks` 和 `active_order_index`，`busy()` 对二者取并集。卖出订单也应进入可恢复状态或订单索引。 + +### 3.2 刷新券商订单会整体覆盖刚提交的本地锁 + +位置：`py-client/strategy/trend/order.py:72-89,132-135` + +`refresh()` 使用新字典直接替换 `self.lock`。如果下单接口成功后，券商订单明细存在短暂可见性延迟，下一轮刷新会删除刚写入的本地锁。对于没有状态兜底的卖单，这同样可能导致重复委托。 + +建议：刷新时合并尚未超过防抖时限的本地锁；只有券商明确返回终结状态，或本地锁超时且券商持续不可见，才能移除，并记录告警。 + +### 3.3 IPO 不检查业务返回结果就写成功锁 + +位置：`py-client/strategy/ipo/boot.py:37-55` + +当前 `try` 只能捕获 Python 异常。`client.passorder()` 返回 `{"status": "failed"}`、空订单号或其他业务拒绝响应时不会抛异常，代码仍进入 `else` 并写锁文件。 + +影响：券商没有接受申购，本地却永久标记为已申购，造成漏申购。 + +建议：捕获异常之外，还必须校验返回值为字典、`status == "success"` 且 `order_ref` 有效；满足全部条件后才能写锁。 + +### 3.4 IPO 没有券商委托/成交防重 + +位置：`py-client/strategy/ipo/boot.py:31-57` + +当前只检查本地空文件，不查询当日券商委托和成交。下单成功后、写锁前崩溃，或锁目录被清理/切换时，10:00 与 14:00 两次任务可能对同一证券再次提交申购。 + +建议：以账户、交易日、证券代码和 IPO 备注核对 `trade_detail_data("order")` 与 `deals()`；本地锁只能作为快速缓存，不能作为唯一事实来源。 + +### 3.5 趋势开仓仍可能超过单笔预算和账户可用现金 + +位置： + +- `py-client/libs/calc.py:10-12` +- `py-client/strategy/trend/boot.py:127-185` +- `py-client/strategy/trend/open.py:41-74` + +`calc_buy_volume()` 在预算不足一手时仍强制返回 100 股。趋势开仓只检查一次现金比例，没有校验单笔预计金额，也没有在同一轮多个信号之间预留资金。 + +影响：高价股会超过 `buy_value`；多个信号可能同时使用同一份可用现金，发送总额超限的委托。 + +建议：预算不足一手时返回 0；开仓循环维护共享 `remaining_cash`，成功提交后立即扣减预留金额，并保留滑点余量。 + +## 4. P1：重要业务逻辑错误 + +### 4.1 IPO 的 4 项现有测试全部无法进入业务逻辑 + +位置： + +- `py-client/strategy/ipo/boot.py:22` +- `py-client/tests/test_ipo.py:52-136` + +`AutoBuyIpo()` 删除了原有可选时间参数，测试传入固定时间时全部报：`TypeError: AutoBuyIpo() takes 0 positional arguments but 1 was given`。APScheduler 无参调用与保留可选 `now` 并不冲突。 + +测试还要求但当前实现没有满足：成功数量返回、上下文关闭、重启防重、券商记录防重、单只拒绝不影响其他候选。 + +建议：恢复 `now: datetime | None = None`，内部使用 `now or datetime.now()`；修复生产逻辑后让测试通过，不要删除失败测试。 + +### 4.2 IPO 正常完成时返回 `None`，与 `-> int` 不一致 + +位置：`py-client/strategy/ipo/boot.py:22-57` + +只有禁用和非交易时间返回 0；完成候选循环后没有返回值，也没有统计成功数。 + +建议：维护 `success_count`，所有出口都返回整数。 + +### 4.3 IPO 资源关闭仍不具备异常安全性 + +位置：`py-client/strategy/ipo/boot.py:31-57` + +`client.close()` 只在整个流程正常走到末尾时执行。`ipo_data()`、候选字段读取、锁写入等任何异常都会跳过关闭。单只下单异常虽然被捕获，但仅记录普通 info，丢失堆栈和具体原因。 + +建议：使用 `with Client(...) as client:`；单只异常用 `logging.exception()` 并继续其他候选。 + +### 4.4 IPO 只判断工作日，不判断真实交易日 + +位置： + +- `py-client/strategy/ipo/boot.py:27` +- `py-client/libs/calc.py:5-7` + +法定节假日或临时休市仍会进入申购。模块中的 `TRADING_CALENDAR_SYMBOL` 未使用。 + +建议：用 SDK `trading_dates()` 确认当天交易日；查询失败时安全跳过。 + +### 4.5 同证券同方向多笔委托仍会互相覆盖 + +位置：`py-client/strategy/trend/order.py:74-89,132-135` + +`data` 和 `lock` 都以 `SIDE-code` 为唯一键。两笔相同证券、相同方向的订单只保留最后一笔。 + +定向复现：传入两笔 `SELL-A`，刷新后 `len(data) == 1`。 + +影响：被覆盖订单无法被撤销、展示或独立对账。 + +建议：订单数据以系统委托号为主键，另建 `(side, code) -> set[order_id]` 活动索引。 + +### 4.6 撤单状态集合与活动状态集合不一致 + +位置：`py-client/strategy/trend/order.py:15,97-104` + +`BUSY_STATUSES` 包含 `48` 和 `55`，撤单逻辑只处理 `49`～`52`。如果 `48/55` 也是可继续成交且可撤的状态，它们永远不会被超时撤销；同时其锁仍可能在 180 秒后被删除。 + +建议：建立单一、有文档依据的状态映射，分别定义“活动”“可撤”“终结”，不要在不同函数中散落不一致的魔法集合。 + +### 4.7 撤单请求会每轮重复发送 + +位置：`py-client/strategy/trend/order.py:91-104` + +默认 10 秒即触发撤单，主循环每 30 秒刷新一次；只要券商仍返回相同活动状态，每轮都会再次调用 `cancel_by_id()`，没有撤单中、本地冷却或最大重试次数。 + +建议：记录最近撤单请求时间和结果；处于撤单中的订单采用退避重试，并设置最大次数和告警。 + +### 4.8 状态保存失败后的安全策略仍不统一 + +位置： + +- `py-client/strategy/trend/open.py:59-74` +- `py-client/strategy/trend/positions.py:175-183` + +开仓保存失败只记录日志并继续；补仓保存失败抛出异常。两种情况都可能已经有真实订单，但恢复数据未落盘。 + +建议：落盘失败后保留内存订单锁并熔断该证券，持续重试保存；完成券商对账前禁止后续自动交易。 + +### 4.9 启动脚本会杀死机器上的全部 Python 进程 + +位置：`run.bat:1-7` + +`taskkill /IM python.exe /F` 不区分 PID 或项目；脚本又没有先切换到 `py-client`，`python main.py` 依赖调用目录。 + +建议：仅管理本项目 PID；使用 `%~dp0` 构造绝对路径；生产启动不要无条件 `git pull`。 + +## 5. P2：冗余、过度验证与维护风险 + +### 5.1 每轮重复查询两次委托明细 + +位置： + +- `py-client/strategy/trend/boot.py:121-125,147-153` +- `py-client/strategy/trend/order.py:72-93` + +`cancel_expired()` 内部先调用一次 `trade_detail_data("order")`，随后 `RunOnce()` 为状态对账再次调用同一接口。30 秒一轮时属于稳定的重复网络请求，两次快照还可能不一致。 + +建议：每轮只获取一次原始订单快照，同时传给订单簿刷新、撤单判断和状态对账。 + +### 5.2 状态对账重复落盘 + +位置：`py-client/strategy/trend/state.py:107-133,135-162` + +`reconcile()` 先调用会保存的 `sync_positions()`，结束时再次保存。每轮至少写两次状态文件，第一份还是未完成订单对账的中间状态。 + +建议：内部同步只更新内存，完整对账成功后一次原子保存。 + +### 5.3 `State.delete()` 保护仍然过宽且静默 + +位置：`py-client/strategy/trend/state.py:93-105` + +虽然已经返回布尔值，但任何 `ING/UNKNOWN` 都只按状态字符串拒绝删除，没有订单证据、日志或人工确认后的强制清理入口。当前内部清理也没有使用返回值记录拒绝原因。 + +建议：把删除许可作为明确的对账结果；为人工确认提供带原因的审计接口。 + +### 5.4 两份 API 文件逐字节重复，正式入口处于删除状态 + +位置： + +- `api/qmt_api_new.py` +- `api/qmt_api_rele.py` +- 当前删除的 `api/QMT_API.py` + +两个新文件 SHA-256 完全相同，正式加载哪个文件不明确。 + +建议：只保留一个正式入口，版本差异交由 Git 管理。 + +### 5.5 IPO 模块存在未使用的旧设计残留 + +位置：`py-client/strategy/ipo/boot.py:5-19` + +`json`、`Any`、`IPO_STRATEGY_NAME`、`IPO_REMARKS`、`IPO_SESSIONS`、`TRADING_CALENDAR_SYMBOL` 当前均未使用。 + +建议：先完成券商对账和交易日逻辑，再删除确认无用的符号。 + +### 5.6 主程序仍有不可达代码和拼写错误 + +位置：`py-client/main.py:13-14,126-127` + +`StartTrend()` 正常情况下永久循环，后面的日志不可达；即使返回，`config.account_config.strateg` 也不存在。`GLOBAL_CONFIG_PATH` 未使用。 + +建议：启动成功日志放在进入循环前；删除不可达代码和无效常量。 + +### 5.7 `replace_existing=True` 在当前内存调度器中是多余防御 + +位置：`py-client/main.py:115-122` + +每次进程启动都创建全新内存调度器，并只注册一次固定任务，不存在同一调度器重复注册路径。该参数无害，但容易暗示存在持久化 job store。 + +建议：没有持久化任务仓库时可删除；未来启用持久化后再明确任务替换策略。 + +## 6. 本轮已确认改善 + +- APScheduler 在北京时间每日 10:00、14:00 触发，不受趋势永久循环阻塞。 +- IPO 单只 `passorder()` 抛异常时不再写锁，并会继续处理后续候选。 +- `dev.yaml` 已显式启用信号列表。 +- 趋势每轮重新读取委托和成交进行状态对账；对账失败时本轮停止自动交易。 +- `ING/UNKNOWN` 已同时进入候选过滤和提交前检查。 +- `State.delete()` 已返回真实删除结果。 +- 趋势新增的 `UNKNOWN` 防重复测试通过；趋势测试共 8 项全部通过。 +- 订单方向锁已经改为 `dict[str, float]`，过期清理本身可工作。 + +## 7. 建议整改顺序与验收门槛 + +1. 先拆分“本地防抖锁”和“券商活动订单索引”，确保活动委托绝不因时间到期而变成不忙。 +2. 修复 IPO 返回值校验、券商防重、可选时间参数、资源关闭和成功计数，使现有 4 项 IPO 测试通过。 +3. 修复开仓单笔和同轮资金预算。 +4. 将订单簿改为系统委托号主键，统一活动/可撤/终结状态表和撤单冷却。 +5. 消除重复网络查询和重复状态写盘。 +6. 最后清理启动脚本、重复 API 文件和死代码。 + +最低验收要求：现有 12 项测试全部通过；新增“活动卖单超过锁超时仍 busy”“券商刷新延迟不清除本地锁”“两笔同方向订单均可撤销”“IPO 业务失败响应不写锁”“锁丢失但券商已有 IPO 委托不重复申购”；最后在模拟账户完成买入、部分成交、撤单、卖出、重启对账闭环。 diff --git a/docs/todo.md b/docs/todo.md index c3d2111..9b6a228 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -1,5 +1,87 @@ +### 4.2 IPO 没有校验真实交易日，只判断周一至周五和盘中时间 + +位置： + +- `py-client/strategy/ipo/boot.py:27` +- `py-client/libs/calc.py:5-7` + +`trading_time()` 只排除周末，法定节假日、临时休市仍会进入申购逻辑。模块中已有 `TRADING_CALENDAR_SYMBOL` 常量，但没有实际查询交易日接口。 + +建议：调用 SDK 的 `trading_dates()` 验证当天交易日；接口失败时按安全策略跳过，不应猜测为交易日。 + +### 4.8 启动脚本仍会强制终止机器上的所有 Python 进程 + +位置：`run.bat:1-7` + +`taskkill /IM python.exe /F` 不区分项目和 PID；`python main.py` 又依赖调用时工作目录。它可能杀死无关任务后仍因找不到根目录下的 `main.py` 而启动失败。 + +建议：只管理本项目 PID；脚本先切换至 `%~dp0py-client`；生产启动不要无条件 `git pull`。 + +## 5. P2：代码冗余与过度验证 + +### 5.1 两份 API 文件逐字节重复，正式入口被删除 + +位置： + +- `api/qmt_api_new.py` +- `api/qmt_api_rele.py` +- 当前处于删除状态的 `api/QMT_API.py` + +两个新文件 SHA-256 相同，属于完整重复；原正式文件被删除，启动和部署入口不清晰。继续维护会造成修复只落在其中一份的风险。 + +建议：保留唯一正式文件，版本差异交给 Git 管理。 + +### 5.2 IPO 模块保留大量未使用的设计残留 + +位置：`py-client/strategy/ipo/boot.py:5-19` + +`json`、`Any`、`IPO_STRATEGY_NAME`、`IPO_REMARKS`、`IPO_SESSIONS`、`TRADING_CALENDAR_SYMBOL` 当前均未使用。它们与测试要求一起表明完整对账实现被缩减，但残留符号没有同步整理。 + +建议：先恢复交易日、券商对账和统计逻辑，再清除确认无用的符号，不要只做表面删减。 + +### 3.5 开仓金额可能超过配置值，同轮多信号会重复使用现金 + +位置： + +- `py-client/libs/calc.py:10-12` +- `py-client/strategy/trend/boot.py:119-164` +- `py-client/strategy/trend/open.py:41-70` + +`calc_buy_volume()` 在预算不足一手时仍强制返回 100 股。趋势开仓仅检查一次现金比例，没有校验单笔预计金额，也没有为同一轮后续信号扣减已提交订单占用的资金。 + +影响：高价股单笔超出 `buy_value`；多信号集中出现时可能发送总额超过可用资金的委托。 + +建议：不足一手返回 0；开仓入口维护本轮共享 `remaining_cash`，成功提交后立即预留资金，并考虑价格滑点。 + +### 3.2 IPO 仅依赖本地空文件防重，锁丢失或落盘失败会重复申购 + +位置： + +- `py-client/strategy/ipo/boot.py:37-51` +- `py-client/libs/lockfile.py:9-18` + +当前实现没有查询 `trade_detail_data("order")` 或 `deals()`，也没有按账户、交易日和申购备注对账。本地锁文件被清理、数据目录切换、写入失败或下单成功后进程崩溃时，下一次任务会再次提交同一新股。 + +影响：产生重复申购请求；实际结果取决于券商拦截，不能把安全性寄托在券商拒单上。 + +建议：本地记录只能作为快速幂等缓存，最终防重必须以“账户 + 交易日 + 证券代码”的券商委托/成交记录为准；下单前后二次核对。 + +### 3.4 开仓可能超过单笔配置金额，多信号还会重复使用同一份现金 + +位置： + +- `py-client/libs/calc.py:10-12` +- `py-client/strategy/trend/boot.py:119-164` +- `py-client/strategy/trend/open.py:41-70` + +`calc_buy_volume()` 使用 `max(1, floor(...)) * 100`。当 `buy_value < price * 100` 时，它不是返回 0，而是强制购买 100 股，实际金额必然超过 `buy_value`。趋势开仓只检查一次总账户现金比例，没有检查单笔预计金额是否小于可用现金，也没有在同一轮多个信号之间扣减已预留资金。 + +影响：高价股或同轮多信号可能导致单笔超预算、连续发送超过可用资金的委托，产生券商拒单或非预期仓位。 + +建议：预算不足一手时返回 0；像补仓路径一样维护本轮 `remaining_cash`；每次成功提交后立即预留 `price * volume`，并给价格滑点留安全余量。 + ### 2.1 每次启动都会无条件执行新股申购 整改状态：已于 2026-08-29 按本节方案完成，新增 4 项专项测试。 diff --git a/py-client/etc/dev.yaml b/py-client/etc/dev.yaml index 59933ba..b5b120a 100644 --- a/py-client/etc/dev.yaml +++ b/py-client/etc/dev.yaml @@ -6,6 +6,7 @@ loss_trigger_pct: -30 grid_step_pct: 1 min_profit_pct: 2 strategy: trend +signal_allow: ["morning","tail","arbitrage"] enable_loss_add_position: True enable_auto_ipo: True excluded_codes: diff --git a/py-client/main.py b/py-client/main.py index 907aba0..13f9efe 100644 --- a/py-client/main.py +++ b/py-client/main.py @@ -2,10 +2,9 @@ # -*- coding: utf-8 -*- import logging -from logging.handlers import TimedRotatingFileHandler import os import sys -import schedule +from apscheduler.schedulers.background import BackgroundScheduler import config from dataclasses import dataclass import yaml @@ -98,7 +97,7 @@ def wait_for_any_key() -> None: def main() -> int: try: if not require_windows(): - log.error("本程序仅支持 Windows 环境运行") + logging.error("本程序仅支持 Windows 环境运行") return 1 if not check_single_instance(PROJECT_ROOT): return 1 @@ -108,10 +107,21 @@ def main() -> int: raise RuntimeError("配置尚未加载，请先调用 config.load()") wait_for_qmt_api() - # 自动打新与主策略隔离；申购服务失败不能阻止趋势策略启动。 - schedule.every().day.at("10:00").do(AutoBuyIpo) - schedule.run_pending() - logging.info("IPO 自动打新启动成功") + # 后台调度不受趋势策略永久循环阻塞；同一时刻最多执行一个实例。 + scheduler = BackgroundScheduler( + timezone="Asia/Shanghai", + job_defaults={"coalesce": True, "max_instances": 1}, + ) + scheduler.add_job( + AutoBuyIpo, + trigger="cron", + hour="10,14", + minute=0, + id="auto_buy_ipo", + replace_existing=True, + ) + scheduler.start() + logging.info("IPO 自动打新定时任务已启动：每日 10:00、14:00") STRATEGIES[config.account_config.strategy].start_strategy() logging.info("%s 策略启动成功",config.account_config.strateg) diff --git a/py-client/requirements.txt b/py-client/requirements.txt index 2179c67..bef0bdb 100644 --- a/py-client/requirements.txt +++ b/py-client/requirements.txt @@ -1,2 +1,3 @@ httpx>=0.27,<1 PyYAML>=6.0 +APScheduler>=3.10,<4 diff --git a/py-client/sdk/__init__.py b/py-client/sdk/__init__.py index 6ea2685..f442d47 100644 --- a/py-client/sdk/__init__.py +++ b/py-client/sdk/__init__.py @@ -11,4 +11,4 @@ class Client(AccountMixin, DataMixin, TradeMixin, MiscMixin, _HTTPClient): """big-qmt 同步 HTTP 客户端。""" -__all__ = ["Client", "APIError", "BusinessError", "OP_BUY", "OP_SELL", "ORDER_TYPE_VOLUME", "PR_TYPE_LATEST", "QUICK_TRADE_NOW"] +__all__ = ["Client", "APIError", "BusinessError", "OP_BUY", "OP_SELL", "ORDER_TYPE_VOLUME", "PR_TYPE_LATEST", "QUICK_TRADE_NOW", "ORDER_SIDE_BY_OFFSET", "OrderItem", "PositionItem", "parse_order"] diff --git a/py-client/sdk/account.py b/py-client/sdk/account.py index ccc02e5..1fccfd1 100644 --- a/py-client/sdk/account.py +++ b/py-client/sdk/account.py @@ -1,21 +1,26 @@ from typing import Any -from .models import Assets, Position +from .models import Assets, PositionItem class AccountMixin: account_type: str - def _positions(self, path: str) -> tuple[list[str], list[Position]]: - raw = self._post(path, {"account": self.account_type}) or {} - return list(raw), [Position.from_dict(value, code) for code, value in raw.items()] + def _positions(self, path: str) -> tuple[list[str], list[PositionItem]]: + payload = self._post(path, {"account": self.account_type}) or {} + raw = payload.get("data", payload) if isinstance(payload, dict) else payload + if isinstance(raw, list): + positions = [PositionItem.from_trade_detail(item) for item in raw] + return [item.stock_code for item in positions], positions + return list(raw), [PositionItem.from_dict(value, code) for code, value in raw.items()] def positions(self): return self._positions("/api/v2/positions") def holding(self): return self._positions("/api/holding") def assets(self) -> Assets: - data = self._post("/api/v2/assets", {"account": self.account_type}) - return Assets(float(data.get("total", 0)), float(data.get("available", 0))) + payload = self._post("/api/v2/assets", {"account": self.account_type}) or {} + data = payload.get("data", payload) if isinstance(payload, dict) else {} + return Assets.from_dict(data) def total_money(self) -> float: return float(self._post("/api/money/total", {"account": self.account_type}).get("total_money", 0)) def available_money(self) -> float: return float(self._post("/api/money/available", {"account": self.account_type}).get("available_money", 0)) diff --git a/py-client/sdk/models.py b/py-client/sdk/models.py index bf75864..e002a2f 100644 --- a/py-client/sdk/models.py +++ b/py-client/sdk/models.py @@ -1,5 +1,5 @@ from __future__ import annotations - +from datetime import datetime from dataclasses import dataclass, field from typing import Any @@ -12,7 +12,60 @@ def _number(value: Any, kind: type = float) -> Any: @dataclass(slots=True) -class Position: +class OrderItem: + """由 QMT 委托明细解析得到的标准订单记录。""" + id: str + code: str + side: str + remark: str + status: str + created_at: datetime | None + volume: int + local_order_id: str = "" + traded_volume: int = 0 + remaining_volume: int = 0 + exchange_id: str = "" + name: str = "" + price: float = 0.0 + trade_price: float = 0.0 + trade_amount: float = 0.0 + + @classmethod + def from_trade_detail(cls, data: dict[str, Any]) -> "OrderItem": + """从 TradeDetailData 的 QMT 原始字段创建订单。""" + instrument_id = str(data.get("m_strInstrumentID") or "") + exchange_id = str(data.get("m_strExchangeID") or "") + code = ( + f"{instrument_id}.{exchange_id}" + if instrument_id and exchange_id + else instrument_id + ) + remaining_volume = _number(data.get("m_nVolumeTotal"), int) + traded_volume = _number(data.get("m_nVolumeTraded"), int) + remark = str(data.get("m_strRemark") or "") + return cls( + id=str(data.get("m_strOrderSysID") or ""), + code=code, + side={"23": "BUY", "24": "SELL", "48": "BUY", "49": "SELL"}.get( + str(data.get("m_nOffsetFlag")), "" + ), + remark=remark, + status=str(data.get("m_nOrderStatus") or ""), + created_at=_trade_datetime(data), + volume=remaining_volume + traded_volume, + local_order_id=remark.split("|", 1)[0] if remark else "", + traded_volume=traded_volume, + remaining_volume=remaining_volume, + exchange_id=exchange_id, + name=str(data.get("m_strInstrumentName") or ""), + price=_number(data.get("m_dPrice")), + trade_price=_number(data.get("m_dTradePrice")), + trade_amount=_number(data.get("m_dTradeAmount")), + ) + + +@dataclass(slots=True) +class PositionItem: stock_code: str = "" stock_name: str = "" direction: Any = None @@ -31,7 +84,7 @@ class Position: expire_date: str = "" @classmethod - def from_dict(cls, data: dict[str, Any], code: str = "") -> "Position": + def from_dict(cls, data: dict[str, Any], code: str = "") -> "PositionItem": return cls( stock_code=str(data.get("StockCode") or code), stock_name=str(data.get("StockName") or ""), direction=data.get("Direction"), volume=_number(data.get("Volume"), int), @@ -43,12 +96,58 @@ class Position: future_trade_type=data.get("FutureTradeType"), expire_date=str(data.get("ExpireDate") or ""), ) + @classmethod + def from_trade_detail(cls, data: dict[str, Any]) -> "PositionItem": + """从 TradeDetailData/Holding 的 QMT 原始字段创建持仓。""" + instrument_id = str(data.get("m_strInstrumentID") or "") + exchange_id = str(data.get("m_strExchangeID") or "") + stock_code = ( + f"{instrument_id}.{exchange_id}" + if instrument_id and exchange_id + else instrument_id + ) + return cls( + stock_code=stock_code, + stock_name=str(data.get("m_strInstrumentName") or ""), + direction=data.get("m_nDirection"), + volume=_number(data.get("m_nVolume"), int), + open_price=_number(data.get("m_dOpenPrice")), + float_profit=_number(data.get("m_dFloatProfit")), + market_value=_number(data.get("m_dMarketValue")), + stock_holder=str(data.get("m_strStockHolder") or ""), + frozen_volume=_number(data.get("m_nFrozenVolume"), int), + can_use_volume=_number(data.get("m_nCanUseVolume"), int), + on_road_volume=_number(data.get("m_nOnRoadVolume"), int), + yesterday_volume=_number(data.get("m_nYesterdayVolume"), int), + last_price=_number(data.get("m_dLastPrice")), + profit_rate=_number(data.get("m_dProfitRate")), + future_trade_type=data.get("m_eFutureTradeType"), + expire_date=str(data.get("m_strExpireDate") or ""), + ) + @dataclass(slots=True) class Assets: total: float = 0.0 available: float = 0.0 + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "Assets": + """兼容新版 QMT 原始资金字段及旧版简化字段。""" + return cls( + total=_number(data.get("m_dBalance", data.get("total"))), + available=_number(data.get("m_dAvailable", data.get("available"))), + ) + + +def _trade_datetime(data: dict[str, Any]) -> datetime | None: + date = str(data.get("m_strInsertDate") or "") + clock = str(data.get("m_strInsertTime") or "").replace(":", "").zfill(6) + try: + return datetime.strptime(date + clock, "%Y%m%d%H%M%S") + except ValueError: + return None + @dataclass(slots=True) class Tick: diff --git a/py-client/sdk/trade.py b/py-client/sdk/trade.py index 335802c..af19abd 100644 --- a/py-client/sdk/trade.py +++ b/py-client/sdk/trade.py @@ -1,7 +1,9 @@ +from .models import * from typing import Any OP_BUY, OP_SELL = 23, 24 ORDER_TYPE_VOLUME, PR_TYPE_LATEST, QUICK_TRADE_NOW = 1101, 5, 2 +ORDER_SIDE_BY_OFFSET = {"23": "BUY", "24": "SELL", "48": "BUY", "49": "SELL"} class TradeMixin: @@ -53,7 +55,20 @@ class TradeMixin: def pause_task(self, task_id): return self._task("pause", task_id) def resume_task(self, task_id): return self._task("resume", task_id) def do_order(self): return self._post("/api/trade/do_order") - def trade_detail_data(self, datatype): return self._post("/api/trade/trade_detail_data", {"account": self.account_type, "datatype": datatype}).get("data", []) + def trade_detail_data(self, datatype): + datatype = str(datatype).strip().lower() + data = self._post( + "/api/trade/trade_detail_data", + {"account": self.account_type, "datatype": datatype}, + ).get("data", []) + rows = data if isinstance(data, list) else [data] if isinstance(data, dict) else [] + if datatype == "order": + return [OrderItem.from_trade_detail(row) for row in rows] + if datatype == "position": + return [PositionItem.from_trade_detail(row) for row in rows] + if datatype == "account": + return [Assets.from_dict(row) for row in rows] + return data def value_by_order_id(self, order_id, datatype): return self._post("/api/trade/value_by_order_id", {"orderId": order_id, "accountType": self.account_type, "datatype": datatype}).get("data") def last_order_id(self, datatype): return self._post("/api/trade/last_order_id", {"account": self.account_type, "datatype": datatype}).get("last_order_id") def can_cancel_order(self, order_id): return self._post("/api/trade/can_cancel_order", {"orderId": order_id, "accountType": self.account_type}).get("can_cancel") diff --git a/py-client/strategy/ipo/boot.py b/py-client/strategy/ipo/boot.py index 6e55ba7..c6062e8 100644 --- a/py-client/strategy/ipo/boot.py +++ b/py-client/strategy/ipo/boot.py @@ -19,12 +19,12 @@ IPO_SESSIONS = ((time(9, 30), time(11, 30)), (time(13, 0), time(15, 0))) TRADING_CALENDAR_SYMBOL = "000001.SH" -def AutoBuyIpo(now: datetime | None = None) -> int: +def AutoBuyIpo() -> int: """安全执行一次新股申购，返回成功提交的证券数量。""" if not config.account_config.enable_auto_ipo: logging.info("[IPO] 自动申购未启用") return 0 - if not trading_time(): + if not trading_time(datetime.now()): logging.info("[IPO] 非交易时间") return 0 @@ -36,17 +36,22 @@ def AutoBuyIpo(now: datetime | None = None) -> int: result = client.ipo_data("STOCK") for stock in result: - lp = Path(config.global_config.qmt_data_dir/f"{stock}.lock") - if is_lock(lp): + lp = Path(config.global_config.qmt_data_dir)/f"{stock}.lock" + if not is_lock(lp): ipo_price = result[stock]['issuePrice'] # 发行价 maxPurchaseNum = result[stock]['maxPurchaseNum'] # 可申购额度 - client.passorder( - op_type=23, - stock=stock, - volume=maxPurchaseNum, - pr_type=11, - price=ipo_price, - strategy_name="新股申购", - ) - write_lockfile(lp) - + try: + client.passorder( + op_type=23, + stock=stock, + volume=maxPurchaseNum, + pr_type=11, + price=ipo_price, + strategy_name="新股申购", + ) + except Exception: + logging.info("[IPO] %s 申购失败，不写入锁文件", stock) + else: + write_lockfile(lp) + + client.close() diff --git a/py-client/strategy/trend/__init__.py b/py-client/strategy/trend/__init__.py index 7556f2d..e69de29 100644 --- a/py-client/strategy/trend/__init__.py +++ b/py-client/strategy/trend/__init__.py @@ -1,5 +0,0 @@ -from .order import OrderBook, PlaceOrderRequest -from .state import State, StateItem -from .watch import DipWatch -from .open import check_timezone, open_signal -from .positions import manage_positions diff --git a/py-client/strategy/trend/boot.py b/py-client/strategy/trend/boot.py index 515c7ca..9789485 100644 --- a/py-client/strategy/trend/boot.py +++ b/py-client/strategy/trend/boot.py @@ -144,19 +144,26 @@ def RunOnce(run: Runtime, signals) -> None: logging.exception("获取持仓失败") return - active_codes = set(position_codes) - removed_codes = set(run.state.codes) - active_codes + # 每轮使用最新委托和成交恢复状态。查询或落盘失败时禁止继续开仓， + # 避免在订单结果不明确的情况下提交重复买单。 + previous_state_codes = set(run.state.codes) + try: + broker_orders = run.client.trade_detail_data("order") + broker_deals = run.client.deals() + run.state.reconcile(positions, broker_orders, broker_deals) + except Exception: + logging.exception("订单状态对账失败，本轮禁止自动交易") + return + + removed_codes = previous_state_codes - set(run.state.codes) for code in removed_codes: - run.state.delete(code) run.open_watch.forget(code) run.add_watch.forget(code) - if removed_codes: - run.state.save() - # 5. 验证有效开仓信号：排除已有持仓，并按 signal_allow 过滤。 + # 5. 验证有效开仓信号：排除已有持仓和未决订单。 position_code_set = set(position_codes) allow_open = [] - seen_codes = set(position_code_set) + seen_codes = position_code_set | set(run.state.unresolved_codes) for signal in signals: if signal.code not in seen_codes: allow_open.append(signal) diff --git a/py-client/strategy/trend/open.py b/py-client/strategy/trend/open.py index a92e3b6..973d0f3 100644 --- a/py-client/strategy/trend/open.py +++ b/py-client/strategy/trend/open.py @@ -15,6 +15,10 @@ from .state import STATUS_ING, StateItem def open_signal(run, ticks, open_signals) -> None: """逐个验证开仓信号并提交买入委托。""" for item in open_signals: + # 候选生成后状态仍可能发生变化，提交前再次阻止未决订单重复开仓。 + if run.state.has_unresolved_order(item.code): + continue + # 1. 验证信号配置允许开仓的时间区间。 signal_config = run.global_cfg.signals.get(item.signal_key) if signal_config is None or not check_timezone(signal_config.timezone): diff --git a/py-client/strategy/trend/order.py b/py-client/strategy/trend/order.py index a94be70..45afd74 100644 --- a/py-client/strategy/trend/order.py +++ b/py-client/strategy/trend/order.py @@ -8,8 +8,7 @@ from datetime import datetime, timedelta from threading import Lock from typing import Any -# QMT 开平方向字段到本地买卖方向的映射。 -OFFSET_FLAG = {"23": "BUY", "24": "SELL", "48": "BUY", "49": "SELL"} +from sdk import ORDER_SIDE_BY_OFFSET, Client, OrderItem # 表示委托仍在处理、可能继续成交的 QMT 状态。 BUSY_STATUSES = {"48", "49", "50", "51", "52", "55"} @@ -27,56 +26,45 @@ class PlaceOrderRequest: strategy_name: str -@dataclass(slots=True) -class OrderItem: - """从 QMT 委托明细转换得到的本地订单记录。""" - - id: str - code: str - side: str - remark: str - status: str - created_at: datetime | None - volume: int - local_order_id: str = "" - - class OrderBook: """线程安全的活动委托缓存。""" - def __init__(self, timeout_seconds: float = 300) -> None: - self.timeout = timedelta(seconds=timeout_seconds) + def __init__(self, lock_timeout_sec: float = 180, cancel_timeout_sec: float = 10) -> None: + self.lock_timeout_sec = max(0.0, float(lock_timeout_sec)) + self.cancel_timeout_sec = timedelta(seconds=cancel_timeout_sec) self.data: dict[str, OrderItem] = {} - self.index: list[str] = [] - self.lock = Lock() + self.lock: dict[str, float] = {} + self.mutex = Lock() @staticmethod def new_order_id(leg: str) -> str: """生成短订单号，为 QMT 备注中的信号键预留空间。""" return f"zt-{leg[:1]}-{secrets.token_hex(4)}" - def is_lock(self, side: str, code: str) -> bool: - """判断证券在指定买卖方向上是否已经被委托锁定。""" - with self.lock: - return f"{side}-{code}" in self.index - def busy(self, code: str, side: str) -> bool: """判断证券是否存在仍在处理中的同方向委托。""" - with self.lock: + with self.mutex: + self._clear_expired_locks(datetime.now().timestamp()) key = f"{side}-{code}" - order = self.data.get(key) - return key in self.index or bool(order and order.status in BUSY_STATUSES) + return key in self.lock - def refresh(self, client: Any) -> None: + def refresh(self, client: Client) -> None: """从 QMT 刷新当前委托明细和方向索引。""" - parsed_orders = [ - parse_order(row) for row in client.trade_detail_data("order") - ] - with self.lock: + orders = client.trade_detail_data("order") + parsed_orders = [(f"{item.side}-{item.code}", item) for item in orders] + now_timestamp = datetime.now().timestamp() + with self.mutex: self.data = {key: item for key, item in parsed_orders} - self.index = [ - key for key, item in parsed_orders if item.status in BUSY_STATUSES - ] + self.lock = { + key: ( + item.created_at.timestamp() + if item.created_at is not None + else now_timestamp + ) + for key, item in parsed_orders + if item.status in BUSY_STATUSES + } + self._clear_expired_locks(now_timestamp) def cancel_expired(self, client: Any, now: datetime | None = None) -> None: """尝试撤销超过有效期且具有委托编号的订单。""" @@ -87,7 +75,8 @@ class OrderBook: for order in list(self.data.values()): if ( order.created_at is not None - and current - order.created_at > self.timeout + and order.status in {"49", "50", "51", "52"} + and current - order.created_at > self.cancel_timeout_sec and order.id ): client.cancel_by_id(order.id) @@ -107,7 +96,7 @@ class OrderBook: if result.get("status") != "success" or order_ref in {"", "unknown", "none"}: return False - side = OFFSET_FLAG.get(str(request.op), "") + side = ORDER_SIDE_BY_OFFSET.get(str(request.op), "") pending = OrderItem( id=order_ref, code=request.code, @@ -118,61 +107,18 @@ class OrderBook: volume=request.volume, local_order_id=request.order_id, ) - with self.lock: + with self.mutex: key = f"{side}-{request.code}" self.data[key] = pending - if key not in self.index: - self.index.append(key) + self.lock[key] = pending.created_at.timestamp() return True - -def parse_order(row: dict[str, Any]) -> tuple[str, OrderItem]: - """把 QMT 原始委托字段转换为本地订单及其索引键。""" - volume = _as_int(row.get("m_nVolumeTotal")) + _as_int( - row.get("m_nVolumeTraded") - ) - - timestamp = _as_int(row.get("m_nOrderTime")) - if timestamp > 100_000_000_000: - # QMT 某些版本返回毫秒时间戳。 - timestamp /= 1000 - created_at = ( - datetime.fromtimestamp(timestamp) - if timestamp - else _parse_insert_datetime(row) - ) - - item = OrderItem( - id=str(row.get("m_strOrderSysID") or ""), - code=str(row.get("m_strInstrumentID") or ""), - side=OFFSET_FLAG.get(str(row.get("m_nOffsetFlag")), ""), - remark=str(row.get("m_strRemark") or ""), - status=str(row.get("m_nOrderStatus") or ""), - created_at=created_at, - volume=volume, - local_order_id=_local_order_id(str(row.get("m_strRemark") or "")), - ) - return f"{item.side}-{item.code}", item - - -def _as_int(value: Any) -> int: - """安全转换整数，无效值按 0 处理。""" - try: - return int(value or 0) - except (TypeError, ValueError): - return 0 - - -def _parse_insert_datetime(row: dict[str, Any]) -> datetime | None: - """使用委托日期和时间字段构造本地时间。""" - date = str(row.get("m_strInsertDate") or "") - clock = str(row.get("m_strInsertTime") or "").replace(":", "").zfill(6) - try: - return datetime.strptime(date + clock, "%Y%m%d%H%M%S") - except ValueError: - return None - - -def _local_order_id(remark: str) -> str: - """兼容 ``local_order_id|signal_key`` 形式的 QMT 备注。""" - return remark.split("|", 1)[0] if remark else "" + def _clear_expired_locks(self, now_timestamp: float) -> None: + """清理过期方向锁；调用方必须已持有 ``mutex``。""" + expired = [ + key + for key, created_at in self.lock.items() + if now_timestamp - created_at >= self.lock_timeout_sec + ] + for key in expired: + self.lock.pop(key, None) diff --git a/py-client/strategy/trend/positions.py b/py-client/strategy/trend/positions.py index b97d6f2..1955577 100644 --- a/py-client/strategy/trend/positions.py +++ b/py-client/strategy/trend/positions.py @@ -7,7 +7,7 @@ from dataclasses import dataclass from libs.calc import calc_buy_volume, calculate_min_profit_rate from libs.grid_take_profit import GridState -from sdk import OP_BUY, OP_SELL, Position, Tick +from sdk import OP_BUY, OP_SELL, PositionItem, Tick from .order import PlaceOrderRequest from .runtime import Runtime @@ -30,7 +30,7 @@ class TradeDecision: def manage_positions( runtime: Runtime, ticks: dict[str, Tick], - positions: list[Position], + positions: list[PositionItem], market_ok: bool, available: float, ) -> None: @@ -88,7 +88,7 @@ def manage_positions( def handle_profit( runtime: Runtime, - position: Position, + position: PositionItem, tick: Tick, pnl_rate: float, minimum_profit: float, @@ -133,7 +133,7 @@ def handle_profit( def handle_loss( runtime: Runtime, - position: Position, + position: PositionItem, tick: Tick, pnl_rate: float, available: float, diff --git a/py-client/strategy/trend/state.py b/py-client/strategy/trend/state.py index a836c75..4fb0bd9 100644 --- a/py-client/strategy/trend/state.py +++ b/py-client/strategy/trend/state.py @@ -8,7 +8,7 @@ from pathlib import Path from threading import Lock from typing import Iterable -from sdk import Position +from sdk import OrderItem, PositionItem # 委托状态：无操作、处理中、已完成。 @@ -70,6 +70,16 @@ class State: with self.lock: return list(self.items) + @property + def unresolved_codes(self) -> list[str]: + """返回存在处理中或未知订单状态的证券代码快照。""" + with self.lock: + return [ + code + for code, item in self.items.items() + if _has_unresolved_order(item) + ] + def get(self, code: str) -> StateItem: """获取指定证券的状态；不存在时抛出 KeyError。""" with self.lock: @@ -80,12 +90,21 @@ class State: with self.lock: self.items[item.code] = item - def delete(self, code: str) -> None: - """删除证券状态；证券不存在时不报错。""" + def delete(self, code: str) -> bool: + """删除已终结的证券状态，并返回是否实际删除。""" with self.lock: - self.items.pop(code, None) + item = self.items.get(code) + if item is not None and _has_unresolved_order(item): + return False + return self.items.pop(code, None) is not None - def sync_positions(self, positions: Iterable[Position]) -> None: + def has_unresolved_order(self, code: str) -> bool: + """判断证券是否存在必须阻止自动下单的未决订单。""" + with self.lock: + item = self.items.get(code) + return item is not None and _has_unresolved_order(item) + + def sync_positions(self, positions: Iterable[PositionItem]) -> None: """把尚未接管的真实持仓初始化为已完成底仓。 无证券代码、无持仓数量或成本无效的记录会被忽略。同步结束后 @@ -115,29 +134,31 @@ class State: def reconcile( self, - positions: Iterable[Position], - orders: list[dict[str, str]], + positions: Iterable[PositionItem], + orders: list[OrderItem], deals: list[dict[str, str]], ) -> None: """用真实持仓、委托和成交恢复本地状态，不增加持久化字段。""" position_list = list(positions) self.sync_positions(position_list) - active_codes = { + position_codes = { item.stock_code for item in position_list if item.volume > 0 } - for code in list(self.codes): - if code not in active_codes: - self.delete(code) - for code in list(self.codes): item = self.get(code) item.base_status = _reconcile_leg( - item.base_order_id, item.base_status, orders, deals + item.base_order_id, item.base_status, item.base_qty, orders, deals ) item.added_status = _reconcile_leg( - item.added_order_id, item.added_status, orders, deals + item.added_order_id, item.added_status, item.added_qty, orders, deals ) self.set(item) + + # Opening orders normally have no position until their first fill. Order + # reconciliation must therefore happen before stale state is removed. + for code in list(self.codes): + if code not in position_codes: + self.delete(code) self.save() def save(self) -> None: @@ -179,31 +200,70 @@ class State: def _reconcile_leg( local_order_id: str, current_status: str, - orders: list[dict[str, str]], + expected_qty: int, + orders: list[OrderItem], deals: list[dict[str, str]], ) -> str: - if current_status != STATUS_ING or not local_order_id: + if current_status not in {STATUS_ING, STATUS_UNKNOWN} or not local_order_id: return current_status - if any(local_order_id in row.get("m_strRemark", "") for row in deals): - return STATUS_OK - order = next( - ( - row for row in orders - if local_order_id in row.get("m_strRemark", "") - ), - None, - ) + matching_orders = [ + order for order in orders if order.local_order_id == local_order_id + ] + order = matching_orders[-1] if matching_orders else None if order is None: + matching_deals = [ + row for row in deals if _matches_local_order(row, local_order_id) + ] + dealt = sum(_deal_volume(row) for row in matching_deals) + if expected_qty > 0 and dealt >= expected_qty: + return STATUS_OK + if expected_qty <= 0 and matching_deals: + return STATUS_OK return STATUS_UNKNOWN - traded = _as_int(order.get("m_nVolumeTraded")) - status = str(order.get("m_nOrderStatus", "")) - if traded > 0 and status not in {"48", "49", "50", "51", "52", "55"}: + + system_order_id = order.id.strip() + matching_deals = [ + row + for row in deals + if ( + system_order_id + and str(row.get("m_strOrderSysID") or "").strip() == system_order_id + ) + or (not system_order_id and _matches_local_order(row, local_order_id)) + ] + dealt = sum(_deal_volume(row) for row in matching_deals) + traded = max(order.traded_volume, dealt) + ordered = order.volume or expected_qty + status = order.status + if ordered > 0 and traded >= ordered: return STATUS_OK + if status in {"48", "49", "50", "51", "52", "55"}: + return STATUS_ING if status in {"54", "56"}: - return STATUS_CANCELED + return STATUS_UNKNOWN if traded > 0 else STATUS_CANCELED if status in {"57", "58"}: - return STATUS_FAILED - return STATUS_ING + return STATUS_UNKNOWN if traded > 0 else STATUS_FAILED + return STATUS_UNKNOWN + + +def _has_unresolved_order(item: StateItem) -> bool: + return item.base_status in {STATUS_ING, STATUS_UNKNOWN} or item.added_status in { + STATUS_ING, + STATUS_UNKNOWN, + } + + +def _matches_local_order(row: dict[str, str], local_order_id: str) -> bool: + remark = str(row.get("m_strRemark") or "") + return remark.split("|", 1)[0] == local_order_id + + +def _deal_volume(deal: dict[str, str]) -> int: + for key in ("m_nVolume", "m_nTradeVolume", "m_nVolumeTraded"): + volume = _as_int(deal.get(key)) + if volume > 0: + return volume + return 0 def _as_int(value: object) -> int: diff --git a/py-client/tests/test_trend.py b/py-client/tests/test_trend.py index aec5ade..6f68611 100644 --- a/py-client/tests/test_trend.py +++ b/py-client/tests/test_trend.py @@ -6,11 +6,11 @@ from types import SimpleNamespace from unittest.mock import patch from libs.grid_take_profit import GridState, GridTrailingTracker -from sdk import Assets, Position, Tick +from sdk import Assets, PositionItem, Tick from strategy.trend.order import OrderBook, PlaceOrderRequest from strategy.trend.positions import LOSS_TIERS, handle_loss, manage_positions from strategy.trend.boot import RunOnce -from strategy.trend.state import STATUS_OK, State, StateItem +from strategy.trend.state import STATUS_OK, STATUS_UNKNOWN, State, StateItem class FakeClient: @@ -42,7 +42,7 @@ class TrendTests(unittest.TestCase): def test_position_dataclasses_execute_without_type_error(self): with TemporaryDirectory() as directory: state = State.for_strategy(directory, "trend", "A") - position = Position( + position = PositionItem( stock_code="000001.SZ", volume=100, can_use_volume=100, open_price=10, market_value=1000, ) @@ -64,7 +64,7 @@ class TrendTests(unittest.TestCase): self.assertEqual(len(LOSS_TIERS), 2) with TemporaryDirectory() as directory: state = State.for_strategy(directory, "trend", "A") - position = Position(stock_code="A", volume=100, open_price=10, market_value=1000) + position = PositionItem(stock_code="A", volume=100, open_price=10, market_value=1000) state.sync_positions([position]) item = state.get("A") item.added_num = len(LOSS_TIERS) @@ -80,7 +80,7 @@ class TrendTests(unittest.TestCase): def test_loss_tiers_zero_and_one(self): with TemporaryDirectory() as directory: state = State.for_strategy(directory, "trend", "A") - position = Position(stock_code="A", volume=100, open_price=10, market_value=1000) + position = PositionItem(stock_code="A", volume=100, open_price=10, market_value=1000) state.sync_positions([position]) runtime = SimpleNamespace( state=state, account_cfg=SimpleNamespace(buy_value=5000, strategy="trend"), @@ -100,7 +100,7 @@ class TrendTests(unittest.TestCase): def test_reconcile_ing_order_from_deal(self): with TemporaryDirectory() as directory: state = State.for_strategy(directory, "trend", "A") - position = Position(stock_code="A", volume=100, open_price=10) + position = PositionItem(stock_code="A", volume=100, open_price=10) state.set(StateItem("A", base_order_id="local-1", base_status="ING")) state.reconcile( [position], @@ -112,7 +112,9 @@ class TrendTests(unittest.TestCase): def test_low_cash_still_runs_position_management(self): client = SimpleNamespace( assets=lambda: Assets(total=10000, available=10), - positions=lambda: (["A"], [Position(stock_code="A", volume=100, open_price=10)]), + positions=lambda: (["A"], [PositionItem(stock_code="A", volume=100, open_price=10)]), + trade_detail_data=lambda _datatype: [], + deals=lambda: [], full_tick=lambda _codes: {"A": Tick(last_price=11)}, ) runtime = SimpleNamespace( @@ -120,7 +122,11 @@ class TrendTests(unittest.TestCase): account_cfg=SimpleNamespace(min_cash_ratio=0.1), global_cfg=SimpleNamespace(api_host="http://example"), orders=SimpleNamespace(cancel_expired=lambda _client: None), - state=SimpleNamespace(codes=["A"]), + state=SimpleNamespace( + codes=["A"], + unresolved_codes=[], + reconcile=lambda *_args: None, + ), ) with ( patch("strategy.trend.boot.trading_time", return_value=True), @@ -132,6 +138,44 @@ class TrendTests(unittest.TestCase): open_mock.assert_not_called() manage_mock.assert_called_once() + def test_unknown_order_without_position_blocks_reopen(self): + with TemporaryDirectory() as directory: + state = State.for_strategy(directory, "trend", "A") + state.set(StateItem( + "A", + base_order_id="missing-order", + base_qty=100, + base_status=STATUS_UNKNOWN, + )) + state.save() + client = SimpleNamespace( + assets=lambda: Assets(total=10000, available=5000), + positions=lambda: ([], []), + trade_detail_data=lambda _datatype: [], + deals=lambda: [], + full_tick=lambda _codes: {"A": Tick(last_price=10)}, + ) + runtime = SimpleNamespace( + client=client, + account_cfg=SimpleNamespace(min_cash_ratio=0.1), + global_cfg=SimpleNamespace(api_host="http://example"), + orders=OrderBook(), + state=state, + open_watch=SimpleNamespace(forget=lambda _code: None), + add_watch=SimpleNamespace(forget=lambda _code: None), + ) + signal = SimpleNamespace(code="A", signal_key="morning") + with ( + patch("strategy.trend.boot.trading_time", return_value=True), + patch("strategy.trend.boot.market_allow_open", return_value=True), + patch("strategy.trend.boot.open_signal") as open_mock, + patch("strategy.trend.boot.manage_positions"), + ): + RunOnce(runtime, [signal]) + + open_mock.assert_not_called() + self.assertTrue(state.has_unresolved_order("A")) + if __name__ == "__main__": unittest.main()