From e4bb57f0116ccbc6498a6b5f916c6ee912707e31 Mon Sep 17 00:00:00 2001 From: zhangxt8113 <168541198+zhangxt8113@users.noreply.github.com> Date: Tue, 18 Nov 2025 11:49:13 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B6=E7=9B=8A18%?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 当前策略已经去掉 MACD 过滤，只剩下“趋势 + 无持仓”两个条件： 从 60 根历史 K 线里计算 MA20、MA60，要求最新价 price > MA60 且 MA20 > MA60，确认处于上升趋势。 khHas(data, code) 返回 False（上一交易日收盘时没有这只股票）。 满足后 _calc_buy_ratio 按“现金×95%，但单只股票不超过账户 5%”计算买入比例，并通过 generate_signal(..., "buy") 下单。 没有其他指标过滤，买点完全由均线趋势 + 空仓判断决定 只有拖尾止损这一条： 每个持仓都记录自买入以来的最高价 highest，并计算两条保护线：highest × (1 − TRAIL_DROP_PCT) 与 highest − TRAIL_ATR_MULT × ATR，取更高者作为动态止损价。 只要当前价跌破这条拖尾线，就触发卖出。首次触发只减半仓（PARTIAL_STOP_RATIO = 0.5），若已经减半或再次触发，则把剩余仓位全部卖出。 --- strategies/kdj_cross_trailing.kh | 70 ++++++++++++ strategies/kdj_cross_trailing.py | 185 +++++++++++++++++++++++++++++++ 2 files changed, 255 insertions(+) create mode 100644 strategies/kdj_cross_trailing.kh create mode 100644 strategies/kdj_cross_trailing.py diff --git a/strategies/kdj_cross_trailing.kh b/strategies/kdj_cross_trailing.kh new file mode 100644 index 0000000..2203523 --- /dev/null +++ b/strategies/kdj_cross_trailing.kh @@ -0,0 +1,70 @@ +{ + "system": { + "userdata_path": "I:/QMT/userdata_mini" + }, + "run_mode": "backtest", + "account": { + "account_id": "88888888", + "account_type": "STOCK" + }, + "strategy_file": "C:/Users/Administrator/Documents/GitHub/OSkhQuant/strategies/kdj_cross_trailing.py", + "data_mode": "custom", + "backtest": { + "start_time": "20250101", + "end_time": "20251101", + "init_capital": 1000000.0, + "min_volume": 100, + "benchmark": "sh.000300", + "trade_cost": { + "min_commission": 5.0, + "commission_rate": 0.0001, + "stamp_tax_rate": 0.0005, + "flow_fee": 0.0, + "slippage": { + "type": "ratio", + "tick_size": 0.01, + "tick_count": 2, + "ratio": 0.01 + } + }, + "trigger": { + "type": "1d", + "custom_times": [ + "09:30:00" + ], + "start_time": "09:30:00", + "end_time": "15:00:00", + "interval": 300 + } + }, + "data": { + "kline_period": "1d", + "dividend_type": "front", + "fields": [ + "open", + "high", + "low", + "close", + "volume", + "amount", + "settelementPrice", + "openInterest", + "preClose", + "suspendFlag" + ], + "stock_list": [ + "000001.SZ" + ] + }, + "market_callback": { + "pre_market_enabled": false, + "pre_market_time": "08:30:00", + "post_market_enabled": false, + "post_market_time": "15:30:00" + }, + "risk": { + "position_limit": 0.95, + "order_limit": 100, + "loss_limit": 0.1 + } +} diff --git a/strategies/kdj_cross_trailing.py b/strategies/kdj_cross_trailing.py new file mode 100644 index 0000000..6079088 --- /dev/null +++ b/strategies/kdj_cross_trailing.py @@ -0,0 +1,185 @@ +# coding: utf-8 +"""MACD trend strategy with MA filter and layered loss control.""" +import logging +from typing import Dict, List + +from khQuantImport import * # noqa: F401,F403 +from MyTT import MA, ATR + +logger = logging.getLogger("macd_trend_strategy") + +MIN_HISTORY_BARS = 60 +SHORT_EMA = 12 +LONG_EMA = 26 +SIGNAL = 9 +BUY_RATIO = 0.95 +MAX_POSITION_RATIO = 0.05 +ATR_PERIOD = 14 +TRAIL_ATR_MULT = 1.0 +TRAIL_DROP_PCT = 0.03 +PARTIAL_STOP_RATIO = 0.5 +STATE_ATTR = "_macd_trend_state" + + +def _calc_buy_ratio(data: Dict, price: float) -> float: + cash = khGet(data, "cash") or 0.0 + total = khGet(data, "total_asset") or 0.0 + if cash <= 0 or total <= 0 or price <= 0: + return 0.0 + intended = cash * BUY_RATIO + per_stock_cap = total * MAX_POSITION_RATIO + order_cash = min(intended, per_stock_cap) + if order_cash <= 0: + return 0.0 + return max(0.0, min(order_cash / cash, 1.0)) + + +def init(stocks=None, data=None): # pragma: no cover + pass + + +def khHandlebar(data: Dict) -> List[Dict]: + signals: List[Dict] = [] + stock_list = khGet(data, "stocks") or data.get("stocks") or [] + date_num = khGet(data, "date_num") or data.get("date_num") + if not stock_list or not date_num: + return signals + + positions = data.get("__positions__", {}) or {} + position_count = sum( + 1 + for info in positions.values() + if info and info.get("volume", 0) > 0 + ) + logger.info( + "positions on %s: %s", + khGet(data, "date") or khGet(data, "date_str") or date_num, + position_count, + ) + state = _ensure_state(data) + position_states = state.setdefault("positions", {}) + _cleanup_state(position_states, positions) + + try: + history = khHistory( + symbol_list=stock_list, + fields=["close", "high", "low"], + bar_count=MIN_HISTORY_BARS, + fre_step="1d", + current_time=date_num, + fq="pre", + force_download=False, + ) + except Exception as err: # pragma: no cover + logger.warning("failed to load history: %s", err) + return signals + + for code in stock_list: + df = history.get(code) + if df is None or len(df) < MIN_HISTORY_BARS: + continue + try: + close = df["close"].values + except KeyError: + continue + highs = df.get("high").values if "high" in df else None + lows = df.get("low").values if "low" in df else None + if len(close) < 60: + continue + ma20 = MA(close, 20)[-1] + ma60 = MA(close, 60)[-1] + price = float(close[-1]) + if price <= ma60 or ma20 <= ma60: + continue + atr_value = _calc_atr(close, highs, lows) + has_pos = khHas(data, code) + if has_pos: + risk_sells = _apply_loss_controls( + data, + code, + price, + atr_value, + positions.get(code) or {}, + position_states, + ) + if risk_sells: + signals.extend(risk_sells) + continue + + if not has_pos: + ratio = _calc_buy_ratio(data, price) + if ratio <= 0: + continue + reason = f"{code[:6]} trend buy" + signals.extend(generate_signal(data, code, price, ratio, "buy", reason)) + return signals + + +def _ensure_state(data: Dict) -> Dict: + framework = data.get("__framework__") + if framework is not None: + state = getattr(framework, STATE_ATTR, None) + if state is None: + state = {} + setattr(framework, STATE_ATTR, state) + return state + return data.setdefault(STATE_ATTR, {}) + + +def _cleanup_state(position_states: Dict, positions: Dict) -> None: + for code in list(position_states.keys()): + if code not in positions: + position_states.pop(code, None) + + +def _calc_atr(close, highs, lows) -> float: + if close is None or highs is None or lows is None: + return 0.0 + try: + atr_series = ATR(close, highs, lows, ATR_PERIOD) + if atr_series is None or len(atr_series) == 0: + return 0.0 + return float(atr_series[-1]) + except Exception as err: # pragma: no cover + logger.debug("ATR calc failed: %s", err) + return 0.0 + + +def _apply_loss_controls( + data: Dict, + code: str, + price: float, + atr_value: float, + position_info: Dict, + position_states: Dict, +) -> List[Dict]: + entry_price = float(position_info.get("avg_price") or 0.0) + if entry_price <= 0: + return [] + state = position_states.setdefault( + code, {"highest": entry_price, "reduced_once": False} + ) + highest = max(float(state.get("highest", entry_price)), price) + state["highest"] = highest + + reason = "" + trail_by_pct = highest * (1 - TRAIL_DROP_PCT) + trail_by_atr = highest - (TRAIL_ATR_MULT * atr_value if atr_value else 0.0) + trailing_price = max(trail_by_pct, trail_by_atr) + if price <= trailing_price: + reason = f"{code[:6]} trail stop {price:.2f}<={trailing_price:.2f}" + if not reason: + return [] + + reduced_once = state.get("reduced_once", False) + ratio = 1.0 if reduced_once else PARTIAL_STOP_RATIO + if ratio <= 0: + return [] + sell_signals = generate_signal(data, code, price, ratio, "sell", reason) + if not sell_signals: + return [] + if reduced_once or ratio >= 1.0: + position_states.pop(code, None) + else: + state["reduced_once"] = True + return sell_signals