股票交易中的非线性回归模型·综合运用
📐

股票交易中的非线性回归模型·综合运用

(3/3)· 当线性指标失效后,怎样把微分方程与自适应比率接进真实盘口并管住回撤

新手友好 第 3/3 篇

很多人把 RSI、MACD 这类七八十年代的线性工具直接套在分形特征明显的市场上,以为调调参数就能预测下一步。历史回测漂亮、实盘却越漂越远,往往不是行情变了,而是模型假设就错了。这一篇把前面两篇铺好的非线性思路收进可用系统,不再谈虚的。

从咖啡馆里的反向信号到自适应仓位

一套趋势模型曾在欧元兑美元看似平稳上扬时弹出空单信号,人为直觉觉得荒谬,但40分钟后欧元跌了35个点——模型抓到的是价格结构里的微变,肉眼很难察觉。这条经验指向一个铁律:别跟系统争辩,尤其外汇和贵金属这种高波动品种,人为情绪往往是亏损源头。 错过信号和内存故障乱开仓的事故,逼出一个轻量警报模块。关键在 _emergency_notification:遇到内存异常或连环开仓,系统发短信并自动停交易,等人工介入才恢复,避免账户被单边扫穿。 仓位管理上,固定0.1手像穿芭蕾鞋走钢丝。后来改成按余额和回撤自适应:基础风险取余额1%,回撤超5%风险减半,再叠ATR做波动率修正。_normalize_volume最麻烦——不同经纪商最小步长不同,有的允0.010手,有的只收整数手,得逐家配参数。 高波动日(美联储讲话、黑五之类)价格像醉酒乱撞。直接关系统太粗暴,改接经济日历API:重要事件前30分钟自动隐身、后30分钟重启,等于给资金加了道缓冲阀。

MQL5 / C++
<span class="keyword">def</span> notify_signal(self, signal_type, message):
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">try</span>:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Format the message</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;timestamp = class="type">class="kw">datetime.now().strftime(<span class="class="type">class="kw">string">&class="macro">#x27;%Y-%m-%d %H:%M:%S&class="macro">#x27;</span>)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;formatted_msg = <span class="class="type">class="kw">string">f"[<span class="subst">{timestamp}</span>] <span class="subst">{signal_type}</span>: <span class="subst">{message}</span>"</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Send to Telegram</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">if</span> self.use_telegram <span class="keyword">and</span> self.telegram_token:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;self.telegram_bot.send_message(
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;chat_id=self.telegram_chat_id,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;text=formatted_msg,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;parse_mode=<span class="class="type">class="kw">string">&class="macro">#x27;HTML&class="macro">#x27;</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Local logging</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">with</span> <span class="built_in">open</span>(self.log_file, <span class="class="type">class="kw">string">&class="macro">#x27;a&class="macro">#x27;</span>, encoding=<span class="class="type">class="kw">string">&class="macro">#x27;utf-class="num">8&class="macro">#x27;</span>) <span class="keyword">as</span> f:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;f.write(<span class="class="type">class="kw">string">f"<span class="subst">{formatted_msg}</span>\n"</span>)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Check critical signals</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">if</span> signal_type <span class="keyword">in</span> [<span class="class="type">class="kw">string">&class="macro">#x27;ERROR&class="macro">#x27;</span>, <span class="class="type">class="kw">string">&class="macro">#x27;MARGIN_CALL&class="macro">#x27;</span>, <span class="class="type">class="kw">string">&class="macro">#x27;CRITICAL&class="macro">#x27;</span>]:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;self._emergency_notification(formatted_msg)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">except</span> Exception <span class="keyword">as</span> e:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># If the notification failed, send the message to the console at the very least</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="built_in">print</span>(<span class="class="type">class="kw">string">f"Error sending notification: <span class="subst">{str(e)}</span>\n<span class="subst">{formatted_msg}</span>"</span>)
<span class="keyword">def</span> calculate_position_size(self):
&nbsp;&nbsp;&nbsp;&nbsp;<span class="class="type">class="kw">string">"""Calculating the position size taking into account volatility and drawdown"""</span>
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">try</span>:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Take the total balance and the current drawdown</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;account_info = mt5.account_info()
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;current_balance = account_info.balance
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;drawdown = (account_info.equity / account_info.balance - <span class="number">class="num">1</span>) * <span class="number">class="num">100</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Basic risk - class="num">1% of the deposit</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;base_risk = current_balance * <span class="number">class="num">0.01</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Adjust for current drawdown</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">if</span> drawdown &lt; -<span class="number">class="num">5</span>:&nbsp;&nbsp;<span class="comment"># If the drawdown exceeds class="num">5%</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;risk_factor = <span class="number">class="num">0.5</span>&nbsp;&nbsp;<span class="comment"># Slash the risk in half</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">else</span>:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;risk_factor = <span class="number">class="num">1</span> - <span class="built_in">abs</span>(drawdown) / <span class="number">class="num">10</span>&nbsp;&nbsp;<span class="comment"># Smooth decrease</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Take into account the current ATR</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;atr = self.calculate_atr()

「仓位与市况的风控闸门」

这段逻辑把「下多少手」和「现在能不能做」拆成了两道独立闸门。前者先取点值,用 (base_risk * risk_factor) 除以 (atr * pip_value) 算出原始手数,再交给 _normalize_volume 按经纪商可开手数取整;异常时退回 0.1 手保底,并抛出错误信号。 市况检查更硬:先用 _is_high_impact_news_time 拦掉高影响新闻窗口;再用 ATR(5) 对比 ATR(20),若短周期波动超过常态两倍直接放弃交易,并回传具体数值便于复盘。 最后一道是点差过滤,当前 spread 大于 max_allowed_spread 同样不放行。外汇与贵金属杠杆高,这类双重拦截能显著降低滑点与事件跳空的概率,建议直接把代码丢进 MT5 的 Python 桥或重写进 EA 实跑验证。

MQL5 / C++
    pip_value = self.get_pip_value()
    
    # Volume calculation rounded to available lots
    raw_volume = (base_risk * risk_factor) / (atr * pip_value)
    class="kw">return self._normalize_volume(raw_volume)
    
except Exception as e:
    self.notify_signal(&class="macro">#x27;ERROR&class="macro">#x27;, f"Volume calculation error: {str(e)}")
    class="kw">return class="num">0.1  # Minimum safety volume
def check_market_conditions(self):
    """Checking the market status before a deal"""
    # Check the calendar of events
    if self._is_high_impact_news_time():
        class="kw">return False
        
    # Calculate volatility
    current_atr = self.calculate_atr(period=class="num">5)  # Short period
    normal_atr = self.calculate_atr(period=class="num">20)  # Normal period
    
    # Skip if the current volatility is class="num">2+ times higher than the norm
    if current_atr > normal_atr * class="num">2:
        self.notify_signal(
            &class="macro">#x27;INFO&class="macro">#x27;,
            f"Increased volatility: ATR(class="num">5)={current_atr:.5f}, "
            f"ATR(class="num">20)={normal_atr:.5f}"
        )
        class="kw">return False
        
    # Check the spread 
    current_spread = mt5.symbol_info(self.symbol).spread
    if current_spread > self.max_allowed_spread:
        class="kw">return False
        
    class="kw">return True

◍ 用 ATR 给止损留活路

早期用固定止损跑实盘,一周就被噪音扫掉一半仓位,市场专治各种想当然。后来改用平均真实波幅(ATR)动态定止损:波动大就放宽,平静期就收紧,系统对行情的适应能力明显上来。 入场逻辑尽量做减法。新手最容易犯的是信号频发——买、卖、再买来回打脸。加一道 15 分钟交易间隔和持仓过滤,这种可笑的简单改动反而救了回测曲线。 风险线只有一条死规矩:单笔亏损不超账户 1%。按 ATR 50 点算,舒适仓位是 0.2 手;欧盘 EURUSD 真实流动时表现最好,重要新闻前停手更划算,模型跟不上那种混乱。 下面这段开仓函数把 ATR 直接嵌进止损止盈:BUY 时 SL 放在 entry - atr,TP 只给 atr/3,赔率偏保守但触发逻辑清晰。 def open_position(self): try: atr = self.calculate_atr() # 取当前 ATR 值,用于动态间距 predicted_price = self.get_model_prediction() # 模型给出的预测价 current_price = mt5.symbol_info_tick(self.symbol).ask # 实时卖一价(做多参照) signal = "BUY" if predicted_price > current_price else "SELL" # 预测价高于现价看多,否则看空 if signal == "BUY": entry = mt5.symbol_info_tick(self.symbol).ask # 多单入场用 ask sl_level = entry - atr # 止损退 atr 点,随波动放大 tp_level = entry + (atr / 3) # 止盈仅 atr 三分之一,快进思路 else: entry = mt5.symbol_info_tick(self.symbol).bid # 空单入场用 bid sl_level = entry + atr tp_level = entry - (atr / 3) request = { "action": mt5.TRADE_ACTION_DEAL, # 市价成交 "symbol": self.symbol, "volume": self.lot_size, "type": mt5.ORDER_TYPE_BUY if signal == "BUY" else mt5.ORDER_TYPE_SELL, "price": entry, "sl": sl_level, "tp": tp_level, "deviation": 20, # 允许 20 点滑点 "magic": 234000, "comment": f"pred:{predicted_price:.6f}", # 备注写预测价便于复盘 "type_filling": mt5.ORDER_FILLING_FOK, # 成交或全撤 } result = mt5.order_send(request) if result.retcode != mt5.TRADE_RETCODE_DONE: raise ValueError(f"Error opening position: {result.retcode}") print(f"Position opened {signal}: price={entry:.5f}, SL={sl_level:.5f}, " f"TP={tp_level:.5f}, ATR={atr:.5f}") return result.order except Exception as e: print(f"Position opening failed: {str(e)}") return None 外汇与贵金属杠杆高,ATR 止损也只是降低噪音误杀概率,不等于控住所有跳空风险。把上面代码丢进 MT5 的 Python 环境,改 symbol 和 lot_size 就能看自己的品种怎么反应。

MQL5 / C++
def open_position(self):
    try:
        atr = self.calculate_atr()
        predicted_price = self.get_model_prediction()
        current_price = mt5.symbol_info_tick(self.symbol).ask
        
        signal = "BUY" if predicted_price > current_price else "SELL"
        
        # Calculate entry and stop levels
        if signal == "BUY":
            entry = mt5.symbol_info_tick(self.symbol).ask
            sl_level = entry - atr   
            tp_level = entry + (atr / class="num">3)
        else:
            entry = mt5.symbol_info_tick(self.symbol).bid
            sl_level = entry + atr
            tp_level = entry - (atr / class="num">3)
        # Send an order
        request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": self.symbol,
            "volume": self.lot_size,
            "type": mt5.ORDER_TYPE_BUY if signal == "BUY" else mt5.ORDER_TYPE_SELL,
            "price": entry,
            "sl": sl_level,
            "tp": tp_level,
            "deviation": class="num">20,
            "magic": class="num">234000,
            "comment": f"pred:{predicted_price:.6f}",
            "type_filling": mt5.ORDER_FILLING_FOK,
        }
        result = mt5.order_send(request)
        if result.retcode != mt5.TRADE_RETCODE_DONE:
            raise ValueError(f"Error opening position: {result.retcode}")
            
        print(f"Position opened {signal}: price={entry:.5f}, SL={sl_level:.5f}, "
              f"TP={tp_level:.5f}, ATR={atr:.5f}")
        
        class="kw">return result.order
    except Exception as e:
        print(f"Position opening failed: {str(e)}")
        class="kw">return None

把数学交给Python,把下单交给MQL5

把整套数学框架硬塞进MQL5是个坑。Python配科学计算库做参数寻优顺手,MQL5跑交易逻辑才高效,没必要拿螺丝刀当锤子使。折中方案是用Python跑Nelder-Mead迭代挑出七个系数,MQL5只负责按系数算预测值和下单。 这七个系数就是模型核心:0.2752466、0.01058082、0.55162082、0.03687016、0.27721318、0.1483476、0.0008025,背后是数周优化和数千次迭代。预测方程几乎原样搬进MQL5,靠t-1、t-2两个价格点的线性、二次、差值、正弦项加常数拼出结果。 锁仓账户回测自2015年起收益约40%,回撤仅0.82%,月均超4%,期间累计成交量7800手、额外贡献约1.5%收益。测试用的最近一年数据做比率优化,这类系统最好零杠杆跑,收益略高于债基和美元存款就够。 别把正态当圣经 DCA时间加权仓位平均在这套EA里是核心机制,外汇和贵金属品种上风险极高。锁仓模式保守参数回测漂亮,但实盘可能一次性清空账户,碰这类策略先想清楚爆仓距离。 让小布替你跑这套 新K线才触发逻辑、目标盈利平掉所有仓、可同持多仓加量——这些都在三百行内写完。想换参数组合,Python重算后改g_coeffs数组即可,交易模块不用动。

MQL5 / C++
class="type">class="kw">double g_coeffs[class="num">7] = {class="num">0.2752466, class="num">0.01058082, class="num">0.55162082, class="num">0.03687016, 
                              class="num">0.27721318, class="num">0.1483476, class="num">0.0008025};
class="type">class="kw">double GetPrediction(class="type">class="kw">double price_t1, class="type">class="kw">double price_t2)
{
   class="kw">return g_coeffs[class="num">0] * price_t1 +                      class=class="str">"cmt">// Linear t-class="num">1
          g_coeffs[class="num">1] * MathPow(price_t1, class="num">2) +           class=class="str">"cmt">// Quadratic t-class="num">1
          g_coeffs[class="num">2] * price_t2 +                      class=class="str">"cmt">// Linear t-class="num">2
          g_coeffs[class="num">3] * MathPow(price_t2, class="num">2) +           class=class="str">"cmt">// Quadratic t-class="num">2
          g_coeffs[class="num">4] * (price_t1 - price_t2) +          class=class="str">"cmt">// Price change
          g_coeffs[class="num">5] * MathSin(price_t1) +              class=class="str">"cmt">// Cyclic
          g_coeffs[class="num">6];                                  class=class="str">"cmt">// Constant
}
class="type">void OpenPosition(class="type">bool buy_signal, class="type">class="kw">double lot)
{
   class="type">MqlTradeRequest request;
   class="type">MqlTradeResult result;
   ZeroMemory(request);
   
   request.action = TRADE_ACTION_DEAL;
   request.symbol = Symbol();
   request.volume = lot;
   request.type = buy_signal ? ORDER_TYPE_BUY : ORDER_TYPE_SELL;
   request.price = buy_signal ? SymbolInfoDouble(Symbol(), SYMBOL_ASK) : 
                               SymbolInfoDouble(Symbol(), SYMBOL_BID);
   class=class="str">"cmt">// ... other parameters
}
if(total_profit >= ProfitTarget)
{
   CloseAllPositions();
   class="kw">return;
}
class="type">bool isNewBar() {
   class="type">class="kw">datetime lastbar_time = class="type">class="kw">datetime(SeriesInfoInteger(Symbol(), 
                                 PERIOD_CURRENT, SERIES_LASTBAR_DATE));
   if(last_time == class="num">0) {
      last_time = lastbar_time;
      class="kw">return(false);
   }
   if(last_time != lastbar_time) {
      last_time = lastbar_time;
      class="kw">return(true);
   }
   class="kw">return(false);
}

「最后一句大实话」

附件里那两个文件就是这套思路的落脚点:MarketSolver.py 是 12.61 KB 的 Python 选优与在线交易脚本,MarketSolver.mq5 是 16.62 KB 的 MQL5 实盘 EA,前者挑比率、后者跑交易,二者靠同一组比率参数咬合。 评论区有人拿原始 py 接了 Qt5 界面筛货币对做实验,也有人坚持零栏 tick 加第一根 K 线就够了——说明落地姿势本就多样,但都没脱离「选优+执行」这条主线。 系统永远没有完美态,外汇和贵金属又是高杠杆高风险场地,你能做的是让 EA 足够聪明去抓概率优势,又足够简单不在极端行情里自爆。深夜里写到这里,剩下的路,开 MT5 把那 16.62 KB 的 EA 拖进图表自己跑一遍最实在。

把回测漂移交给小布盯盘
非线性模型在样本外退化的过程,小布盯盘的 AIGC 已内置了漂移监测,打开对应品种页就能看到预测带与实际价的分离度,你只管判断是否降仓。

常见问题

思路可迁移,但外汇贵金属杠杆高、跳空频繁,属高风险品种,非线性假设同样可能失效,需先做样本外验证再考虑仓位。
倾向用预测区间而非单点,止损放在区间下沿之外一段缓冲,可能减少被噪音扫掉的概率,但黑天鹅下一样会触达。
目前小布内置的是漂移与诊断视图,模型本体仍需你在 MT5 部署;把重复劳动交给小布,你专注决策即可。
差在实时性与数据接口,MQL5 直接吃报价事件,延迟低,但数值库弱,复杂微分求解常要预计算后传参。
先按波动分级切段,再在每个段内拟合局部非线性项,可能比全局单模型更稳,不过要计算开销上升的代价。