用于预测金融时间序列的生物神经元(基础篇)
📘

用于预测金融时间序列的生物神经元(基础篇)

第 1/2 篇

「用霍奇金-赫胥黎模型抓市场信息流」

传统神经网络和遗传算法在金融序列上常有“机械感”过强的短板,难以还原由活人交易行为堆叠出的波动纹理。把视角切到生物侧,直接用诺贝尔奖的霍奇金-赫胥黎模型去模拟神经元放电,反而更贴近市场里信息沿参与者链条扩散的真实样子。 霍奇金-赫胥黎模型原本刻画的是细胞层面神经冲动的产生与传导。映射到盘面上,神经元靠突触传电信号,和交易员靠报单、撤单传信息几乎同构;哪根“突触”先兴奋,哪块流动性就先动。 我们组在经典模型里塞进了一层类等离子体组件:把整个网络看成泡在市场信息“等离子体”里的动态系统,单个神经元不光走硬连接,还能靠自身场去扰动别的神经元。这样一些传统算法漏掉的弱相关,可能被显形。外汇与贵金属杠杆高、跳空频繁,这类模型只提供概率倾向,不等于能稳吃波段。 这篇后续会拆系统架构、运行逻辑和多品种实测。先记住一点:生物启发模型给的是新视角,不是预言机,开 MT5 接数据回测前先想清楚你的样本外风险。

把神经元放电搬进报价流

霍奇金-赫胥黎模型原本描述神经元靠钠、钾离子跨膜运动产生电脉冲,核心是对离子通道门控状态的微分方程刻画。把它借到交易里,逻辑是把每一笔报价和价格变动转译成‘离子电流’:钠通道激活类比买压涌入,钾通道恢复类比卖压释放,膜电位越过阈值就相当于出现一次可交易的脉冲信号。 下面这段 Python 类(注意:原文以 MQL5 语境给出但代码实为 Python,保留原样供你在 MT5 的 Python 桥或本地回测框架里对照)给出了单个神经元的最小骨架。初始膜电位 -65.0 mV 对应行情的‘静默基线’,m=0.05、h=0.6、n=0.32 是三种通道的初始开闭比,last_spike_time 设为负无穷表示尚未发过脉冲。 离子电流计算里,I_Na 用 m 的三次方乘 h,说明钠通道需要近乎同步激活才放巨量电流;I_K 用 n 的四次方,钾恢复更慢更平滑。plasma_influence 把‘上次尖峰到现在的时间’做指数衰减,再乘一个 market_correlation——这直接映射‘一条旧新闻对现价影响逐渐衰减’。 STDP 权重更新段值得在 MT5 里手测:若后突触尖峰晚于前突触(delta_t>0),权重乘 (1+A_plus*e^(-dt/tau)) 走增强;反之走抑制。把 A_plus、A_minus 和 tau 接到 EURUSD 的 M1 回测上,可能看到对突破延续或假突破的识别倾向变化。外汇与贵金属杠杆高,这类生物启发模型只降低主观误判概率,不消除爆仓风险。

MQL5 / C++
<span class="keyword">class</span> HodgkinHuxleyNeuron:
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">def</span> __init__(self):
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;self.V = -<span class="number">class="num">65.0</span>&nbsp;&nbsp;<span class="comment"># Initial resting potential</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;self.m = <span class="number">class="num">0.05</span>&nbsp;&nbsp; <span class="comment"># Activation of sodium channels</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;self.h = <span class="number">class="num">0.6</span>&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Inactivation of sodium channels</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;self.n = <span class="number">class="num">0.32</span>&nbsp;&nbsp; <span class="comment"># Activation of potassium channels</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;self.last_spike_time = <span class="built_in">class="type">class="kw">float</span>(<span class="class="type">class="kw">string">&class="macro">#x27;-inf&class="macro">#x27;</span>)
<span class="keyword">def</span> ion_currents(self, V):
&nbsp;&nbsp;&nbsp;&nbsp;I_Na = self.g_Na * (self.m ** <span class="number">class="num">3</span>) * self.h * (V - self.E_Na)&nbsp;&nbsp;<span class="comment"># Sodium current</span>
&nbsp;&nbsp;&nbsp;&nbsp;I_K = self.g_K * (self.n ** <span class="number">class="num">4</span>) * (V - self.E_K)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <span class="comment"># Potassium current</span>
&nbsp;&nbsp;&nbsp;&nbsp;I_L = self.g_L * (V - self.E_L)&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"># Leakage current</span>
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">class="kw">return</span> I_Na, I_K, I_L
def plasma_influence(self, current_time):
&nbsp;&nbsp;&nbsp;&nbsp;time_since_spike = current_time - self.last_spike_time
&nbsp;&nbsp;&nbsp;&nbsp;influence = self.plasma_strength * np.<span class="functions">exp</span>(-time_since_spike / self.plasma_decay)
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">class="kw">return</span> influence * self.get_market_correlation()
<span class="keyword">def</span> update_synaptic_weights(self, pre_spike, post_spike, weight):
&nbsp;&nbsp;&nbsp;&nbsp;delta_t = post_spike - pre_spike
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">if</span> delta_t &gt; <span class="number">class="num">0</span>:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">class="kw">return</span> weight * (<span class="number">class="num">1</span> + self.A_plus * np.exp(-delta_t / self.tau_plus))
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">else</span>:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">class="kw">return</span> weight * (<span class="number">class="num">1</span> - self.A_minus * np.exp(delta_t / self.tau_minus))
<span class="keyword">class</span> MarketFeatures:
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">def</span> __init__(self, window_size=<span class="number">class="num">20</span>):
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;self.window_size = window_size
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;self.scaler = StandardScaler()
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">def</span> add_price(self, price: <span class="built_in">class="type">class="kw">float</span>, ohlc_data: pd.DataFrame) -&gt; <span>Dict</span>[<span class="built_in">str</span>, <span class="built_in">class="type">class="kw">float</span>]:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;features = {}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Technical indicators</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;features[<span class="class="type">class="kw">string">&class="macro">#x27;sma_10&class="macro">#x27;</span>] = self._calculate_sma(ohlc_data[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>], window=<span class="number">class="num">10</span>)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;features[<span class="class="type">class="kw">string">&class="macro">#x27;ema_20&class="macro">#x27;</span>] = self._calculate_ema(ohlc_data[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>], window=<span class="number">class="num">20</span>)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;features[<span class="class="type">class="kw">string">&class="macro">#x27;rsi&class="macro">#x27;</span>] = self._calculate_rsi(ohlc_data[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>], window=<span class="number">class="num">14</span>)

◍ 把成交量与时序塞进特征向量

这段逻辑先抓两类原始特征:用 10 根 K 线的 tick_volume 算简单移动平均,作为量能基线;再把最近一根 Bar 的小时数与星期几直接写进字典,把时间结构也变成可训练输入。 特征凑齐后一次性 reshape 成 (1, -1) 送进 scaler 做归一化,意味着每次只喂单样本、在线标准化,适合 MT5 逐根 Bar 推演的场景。外汇与贵金属波动受时段切换影响明显,hour 和 day_of_week 这两个字段对欧美盘分界识别有实际参考价值,但高频噪声也可能让过拟合概率上升。 后面接的 BioTradingModel 只是把常规 MLP 的隐藏层塞进 HodgkinHuxleyNeuron 列表,PlasmaField 用 0.95 的衰减率累积神经元活跃度再回灌权重调制——这类生物启发模块在回测里可能改善局部极值逃逸,却还没看到跨品种稳健性的公开数据,建议先拿黄金 M5 小样本跑通再谈放大。

MQL5 / C++
features[&class="macro">#x27;volume_sma&class="macro">#x27;] = self._calculate_sma(ohlc_data[&class="macro">#x27;tick_volume&class="macro">#x27;], window=class="num">10)
features[&class="macro">#x27;hour&class="macro">#x27;] = ohlc_data.index[-class="num">1].hour
features[&class="macro">#x27;day_of_week&class="macro">#x27;] = ohlc_data.index[-class="num">1].dayofweek
class="kw">return self.scaler.fit_transform(np.array(list(features.values())).reshape(class="num">1, -class="num">1))
class BioTradingModel(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(BioTradingModel, self).__init__()
        self.layers = nn.ModuleList([
            nn.Linear(input_size, hidden_size),
            nn.Tanh(),
            nn.Linear(hidden_size, hidden_size),
            nn.Tanh(),
            nn.Linear(hidden_size, output_size)
        ])
        self.bio_neurons = [HodgkinHuxleyNeuron() for _ in range(hidden_size)]
        self.plasma_field = PlasmaField(hidden_size)
class PlasmaField:
    def __init__(self, size):
        self.field_strength = np.zeros(size)
        self.decay_rate = class="num">0.95
    def update(self, neuron_activities):
        self.field_strength = self.field_strength * self.decay_rate
        self.field_strength += neuron_activities
    def get_influence(self, neuron_index):
        class="kw">return np.sum(self.field_strength * np.exp(-self.distance_matrix[neuron_index]))
def train_step(self, inputs, target):
    predictions = self.forward(inputs)
    loss = self.criterion(predictions, target)
    self.optimizer.zero_grad()
    loss.backward()
    for i, neuron in enumerate(self.bio_neurons):
        neuron.update_weights(self.last_spike_times)
        plasma_influence = self.plasma_field.get_influence(i)
        neuron.modulate_weights(plasma_influence)
    self.optimizer.step()
    class="kw">return loss.item()
def calculate_moving_averages(self, prices):
    def sma(window):
        class="kw">return np.convolve(prices, np.ones(window)/window, mode=&class="macro">#x27;valid&class="macro">#x27;)
    def ema(window):
        alpha = class="num">2 / (window + class="num">1)
        kernel = alpha * (class="num">1 - alpha)**np.arange(window)
        class="kw">return np.convolve(prices, kernel[::-class="num">1], mode=&class="macro">#x27;valid&class="macro">#x27;)
    class="kw">return {

「把均线振荡器与波动量能塞进同一张特征表」

这段逻辑把趋势类与摆动类指标拆成三层函数:均线组给快慢双轨,振荡器层算 RSI、动量、随机 K,市场动态层再补布林带与量能剖面。直接拷进 MT5 的 Python 终端或本地回测框架,能立刻看到 10/20 周期 SMA、EMA 与 14 周期 RSI 的同屏输出。 RSI 用 np.diff 取价差后,前 14 根做简单均值种子,之后按 (旧值*13+新值)/14 滚动,和 MT5 内置 RSI 的 Wilder 平滑一致;动量则拿收盘价减 10 根前的滚动值,属于裸动量而非百分比。 布林带取最近 20 根算均值与标准差,上下轨各偏移 2 倍 std,width 用 4*std/sma 做归一化带宽,方便跨品种比波动率。量能剖面里 volume_oscillator 是近 5 均量除近 20 均量减 1 乘 100,读数超 0 可能暗示短期放量。 外汇与贵金属杠杆高、滑点跳空频繁,这类合成特征只反映历史统计关系,实盘信号失效概率不低,上 MT5 验证前先用小仓位或 demo 跑一轮。

MQL5 / C++
            &class="macro">#x27;sma_fast&class="macro">#x27;: sma(class="num">10),   # Fast SMA for class="type">short-term trends 
            &class="macro">#x27;sma_slow&class="macro">#x27;: sma(class="num">20),   # Slow SMA for class="type">long-term trends
            &class="macro">#x27;ema_fast&class="macro">#x27;: ema(class="num">10),   # Exponential MA for fast response
            &class="macro">#x27;ema_slow&class="macro">#x27;: ema(class="num">20)    # Slow EMA for sorting out noise
    }
def calculate_oscillators(self, data):
    def rsi(prices, period=class="num">14):
        delta = np.diff(prices)
        gain = np.where(delta > class="num">0, delta, class="num">0)
        loss = np.where(delta < class="num">0, -delta, class="num">0)
        
        avg_gain = np.mean(gain[:period])
        avg_loss = np.mean(loss[:period])
        
        for i in range(period, len(gain)):
            avg_gain = (avg_gain * class="num">13 + gain[i]) / class="num">14
            avg_loss = (avg_loss * class="num">13 + loss[i]) / class="num">14
            
        rs = avg_gain / avg_loss
        class="kw">return class="num">100 - (class="num">100 / (class="num">1 + rs))
    
    class="kw">return {
        &class="macro">#x27;rsi&class="macro">#x27;: rsi(data[&class="macro">#x27;close&class="macro">#x27;]),
        &class="macro">#x27;momentum&class="macro">#x27;: data[&class="macro">#x27;close&class="macro">#x27;] - np.roll(data[&class="macro">#x27;close&class="macro">#x27;], class="num">10),
        &class="macro">#x27;stoch_k&class="macro">#x27;: self._calculate_stochastic_k(data)
    }
def measure_market_dynamics(self, data):
    def bollinger_bands(prices, window=class="num">20):
        sma = np.mean(prices[-window:])
        std = np.std(prices[-window:])
        class="kw">return {
            &class="macro">#x27;upper&class="macro">#x27;: sma + class="num">2 * std,
            &class="macro">#x27;lower&class="macro">#x27;: sma - class="num">2 * std,
            &class="macro">#x27;width&class="macro">#x27;: class="num">4 * std / sma   # Normalized strip width
        }
    
    def volume_profile(volumes, prices):
        class="kw">return {
            &class="macro">#x27;volume_ma&class="macro">#x27;: np.mean(volumes[-class="num">10:]),
            &class="macro">#x27;volume_trend&class="macro">#x27;: np.corrcoef(volumes[-class="num">20:], prices[-class="num">20:])[class="num">0,class="num">1],
            &class="macro">#x27;volume_oscillator&class="macro">#x27;: (np.mean(volumes[-class="num">5:]) / np.mean(volumes[-class="num">20:]) - class="num">1) * class="num">100
        }
    
    volatility = bollinger_bands(data[&class="macro">#x27;close&class="macro">#x27;])
    volume = volume_profile(data[&class="macro">#x27;volume&class="macro">#x27;], data[&class="macro">#x27;close&class="macro">#x27;])
    class="kw">return {**volatility, **volume}
def normalize_features(self, features: dict) -> dict:
    class AdaptiveNormalizer:
        def __init__(self, window=class="num">100):
            self.window = window
            self.history = {}
            

常见问题

不能直接当指标用,它更适合作为特征提取思路,把报价流类比成神经元放电来识别信息流突变。
将每根K线的收盘价、成交量、波动率拼成定长向量,再喂给轻量模型做状态分类即可。
可以,小布能按你选的品种自动提取均线振荡器与波动量能并合成特征表,省去手工拼接。
看信息流阈值是否被连续触发,若仅单点脉冲而量能未跟进,大概率属于假突破。
容易乱,建议先只做特征向量实验,外汇高风险,别急着拿去实盘下注。