用于预测金融时间序列的生物神经元(基础篇)
「用霍奇金-赫胥黎模型抓市场信息流」
传统神经网络和遗传算法在金融序列上常有“机械感”过强的短板,难以还原由活人交易行为堆叠出的波动纹理。把视角切到生物侧,直接用诺贝尔奖的霍奇金-赫胥黎模型去模拟神经元放电,反而更贴近市场里信息沿参与者链条扩散的真实样子。 霍奇金-赫胥黎模型原本刻画的是细胞层面神经冲动的产生与传导。映射到盘面上,神经元靠突触传电信号,和交易员靠报单、撤单传信息几乎同构;哪根“突触”先兴奋,哪块流动性就先动。 我们组在经典模型里塞进了一层类等离子体组件:把整个网络看成泡在市场信息“等离子体”里的动态系统,单个神经元不光走硬连接,还能靠自身场去扰动别的神经元。这样一些传统算法漏掉的弱相关,可能被显形。外汇与贵金属杠杆高、跳空频繁,这类模型只提供概率倾向,不等于能稳吃波段。 这篇后续会拆系统架构、运行逻辑和多品种实测。先记住一点:生物启发模型给的是新视角,不是预言机,开 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 回测上,可能看到对突破延续或假突破的识别倾向变化。外汇与贵金属杠杆高,这类生物启发模型只降低主观误判概率,不消除爆仓风险。
<span class="keyword">class</span> HodgkinHuxleyNeuron: <span class="keyword">def</span> __init__(self): self.V = -<span class="number">class="num">65.0</span> <span class="comment"># Initial resting potential</span> self.m = <span class="number">class="num">0.05</span> <span class="comment"># Activation of sodium channels</span> self.h = <span class="number">class="num">0.6</span> <span class="comment"># Inactivation of sodium channels</span> self.n = <span class="number">class="num">0.32</span> <span class="comment"># Activation of potassium channels</span> 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): I_Na = self.g_Na * (self.m ** <span class="number">class="num">3</span>) * self.h * (V - self.E_Na) <span class="comment"># Sodium current</span> I_K = self.g_K * (self.n ** <span class="number">class="num">4</span>) * (V - self.E_K) <span class="comment"># Potassium current</span> I_L = self.g_L * (V - self.E_L) <span class="comment"># Leakage current</span> <span class="keyword">class="kw">return</span> I_Na, I_K, I_L def plasma_influence(self, current_time): time_since_spike = current_time - self.last_spike_time influence = self.plasma_strength * np.<span class="functions">exp</span>(-time_since_spike / self.plasma_decay) <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): delta_t = post_spike - pre_spike <span class="keyword">if</span> delta_t > <span class="number">class="num">0</span>: <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)) <span class="keyword">else</span>: <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: <span class="keyword">def</span> __init__(self, window_size=<span class="number">class="num">20</span>): self.window_size = window_size self.scaler = StandardScaler() <span class="keyword">def</span> add_price(self, price: <span class="built_in">class="type">class="kw">float</span>, ohlc_data: pd.DataFrame) -> <span>Dict</span>[<span class="built_in">str</span>, <span class="built_in">class="type">class="kw">float</span>]: features = {} <span class="comment"># Technical indicators</span> 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>) 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>) 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 小样本跑通再谈放大。
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 跑一轮。
&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 = {}