神经网络实践：第一个神经元·综合运用
「基因算子的初始化与扰动逻辑」
这段 Python 片段给出了极简前馈网络的基因组结构：输入维度 5、隐藏层 8 个节点，权重与偏置全部在 [-1,1] 区间均匀撒点，初始 fitness 置 0.0。 突变函数里 mutation_rate 默认 0.1，即每个参数约有 10% 概率被扰动；扰动幅度由 mutation_strength=0.5 控制，加减上限为 ±0.5。隐藏层权重是 8×5 的矩阵，遍历每个元素独立掷骰子决定是否动它。 交叉算子先造一个随机空壳 child，再对 hidden_weights 逐元素以 0.5 概率从父代 1 或父代 2 二选一。注意它只示范了隐藏权重的继承，输出层部分在原文被截断，实盘移植时要补完 output 相关字段的 crossover，否则子代输出恒为随机初值。 在 MT5 里验证这类逻辑，建议先把隐藏层规模压到 3×3，用 EURUSD 的 M1 收盘序列跑 200 代观察过拟合曲线——外汇与贵金属杠杆品种波动剧烈，基因搜索极易在样本内虚假收敛，需以 Out-of-Sample 回测确认概率优势。
def create_random_genome(input_size=class="num">5, hidden_size=class="num">8): class="kw">return { &class="macro">#x27;hidden_weights&class="macro">#x27;: [[random.uniform(-class="num">1, class="num">1) for _ in range(input_size)] for _ in range(hidden_size)], &class="macro">#x27;hidden_bias&class="macro">#x27;: [random.uniform(-class="num">1, class="num">1) for _ in range(hidden_size)], &class="macro">#x27;output_weights&class="macro">#x27;: [random.uniform(-class="num">1, class="num">1) for _ in range(hidden_size)], &class="macro">#x27;output_bias&class="macro">#x27;: random.uniform(-class="num">1, class="num">1), &class="macro">#x27;fitness&class="macro">#x27;: class="num">0.0 } def mutate(genome, mutation_rate=class="num">0.1, mutation_strength=class="num">0.5): for h in range(len(genome[&class="macro">#x27;hidden_weights&class="macro">#x27;])): for i in range(len(genome[&class="macro">#x27;hidden_weights&class="macro">#x27;][h])): if random.random() < mutation_rate: genome[&class="macro">#x27;hidden_weights&class="macro">#x27;][h][i] += random.uniform(-mutation_strength, mutation_strength) for h in range(len(genome[&class="macro">#x27;hidden_bias&class="macro">#x27;])): if random.random() < mutation_rate: genome[&class="macro">#x27;hidden_bias&class="macro">#x27;][h] += random.uniform(-mutation_strength, mutation_strength) for h in range(len(genome[&class="macro">#x27;output_weights&class="macro">#x27;])): if random.random() < mutation_rate: genome[&class="macro">#x27;output_weights&class="macro">#x27;][h] += random.uniform(-mutation_strength, mutation_strength) if random.random() < mutation_rate: genome[&class="macro">#x27;output_bias&class="macro">#x27;] += random.uniform(-mutation_strength, mutation_strength) def crossover(genome1, genome2): child = create_random_genome() # 随机初始值，将被覆盖 for h in range(len(genome1[&class="macro">#x27;hidden_weights&class="macro">#x27;])): for i in range(len(genome1[&class="macro">#x27;hidden_weights&class="macro">#x27;][h])): child[&class="macro">#x27;hidden_weights&class="macro">#x27;][h][i] = (genome1[&class="macro">#x27;hidden_weights&class="macro">#x27;][h][i] if random.random() < class="num">0.5 else genome2[&class="macro">#x27;hidden_weights&class="macro">#x27;][h][i])
遗传交叉与模型落盘的实现细节
这段逻辑把两个基因组按 0.5 概率逐位择一交叉：隐藏层偏置、输出层权重、输出层偏置都走同一套 random.random() < 0.5 的掷硬币规则，子代 fitness 清零后回炉。
进化函数 evolve_population 默认保留 fitness 最高的 5 个基因组（keep_best=5），其余位置用精英里随机抽俩父母交叉+变异补齐，种群规模不变。
训练产物用 save_best_model 写进 best_model.json（indent=2 方便肉眼 diff），下次 load_best_model 先查文件存在否，没有就返 None——直接跳过重训能省掉几十代 EA 迭代。外汇与贵金属行情下用这类模型信号仍属高风险，历史拟合优度不预示实盘概率。
把下面这段直接塞进你的 py 文件就能跑通存读环节：
def save_best_model(genome, filename="best_model.json"): data = { &class="macro">#x27;hidden_weights&class="macro">#x27;: genome[&class="macro">#x27;hidden_weights&class="macro">#x27;], &class="macro">#x27;hidden_bias&class="macro">#x27;: genome[&class="macro">#x27;hidden_bias&class="macro">#x27;], &class="macro">#x27;output_weights&class="macro">#x27;: genome[&class="macro">#x27;output_weights&class="macro">#x27;], &class="macro">#x27;output_bias&class="macro">#x27;: genome[&class="macro">#x27;output_bias&class="macro">#x27;], &class="macro">#x27;fitness&class="macro">#x27;: genome[&class="macro">#x27;fitness&class="macro">#x27;] } with open(filename, &class="macro">#x27;w&class="macro">#x27;) as f: json.dump(data, f, indent=class="num">2) def load_best_model(filename="best_model.json"): if not os.path.exists(filename): class="kw">return None with open(filename, &class="macro">#x27;r&class="macro">#x27;) as f: data = json.load(f) class="kw">return data
◍ 用日切分驱动遗传迭代的回测骨架
这段逻辑把一天当成一代。种群规模设 20，每代保留前 5 个最优基因组进下一代，输入维度 5、隐层 8，属于轻量前馈网络的典型配置，MT5 导出的 CSV 直接能喂。 启动时先尝试 load_best_model()，若本地有存档就把第 0 号基因组的四个权重偏置块整体替换掉，fitness 清零重算；没有就靠 create_random_genome 随机播 20 个种子。 时间解析用 %Y-%m-%dT%H:%M:%S%z 吃带时区的 ISO 串，get_date_from_time_str 只剥出日期。逐行流式读 CSV，row_date 一变就判定「日终」，generation_count 加 1，顺手算平均适应度与最佳适应度打印出来盯进度。 真要验证，把黄金 H1 的报价 CSV 按这个格式丢进 folder_path，跑一轮看 generation_count 是否随自然日累加、best_fitness 是否倾向爬升；外汇与贵金属杠杆高，信号仅作概率参考，实盘前务必小仓校验。
# 超参数 POP_SIZE = class="num">20 KEEP_BEST = class="num">5 input_size = class="num">5 hidden_size = class="num">8 # 创建初始种群或加载最佳模型（如果有的话 population = [create_random_genome(input_size, hidden_size) for _ in range(POP_SIZE)] best_saved = load_best_model() if best_saved is not None: population[class="num">0][&class="macro">#x27;hidden_weights&class="macro">#x27;] = best_saved[&class="macro">#x27;hidden_weights&class="macro">#x27;] population[class="num">0][&class="macro">#x27;hidden_bias&class="macro">#x27;] = best_saved[&class="macro">#x27;hidden_bias&class="macro">#x27;] population[class="num">0][&class="macro">#x27;output_weights&class="macro">#x27;] = best_saved[&class="macro">#x27;output_weights&class="macro">#x27;] population[class="num">0][&class="macro">#x27;output_bias&class="macro">#x27;] = best_saved[&class="macro">#x27;output_bias&class="macro">#x27;] population[class="num">0][&class="macro">#x27;fitness&class="macro">#x27;] = class="num">0.0 # 帮助解析包含时区信息的 ISO8601 日期/时间字符串 def get_date_from_time_str(t_str): dt = class="type">class="kw">datetime.strptime(t_str, "%Y-%m-%dT%H:%M:%S%z") class="kw">return dt.date() csv_files = sorted([f for f in os.listdir(folder_path) if f.endswith(&class="macro">#x27;.csv&class="macro">#x27;)]) # 对于每个基因组，存储最后一个信号和相应的价格（如果有的话） genome_states = [{&class="macro">#x27;last_signal&class="macro">#x27;: None, &class="macro">#x27;last_price&class="macro">#x27;: None} for _ in range(POP_SIZE)] current_date = None generation_count = class="num">0 # 一次处理一个 CSV 文件（逐行流式处理） for csv_file in csv_files: file_path = os.path.join(folder_path, csv_file) with open(file_path, &class="macro">#x27;r&class="macro">#x27;, newline=&class="macro">#x27;&class="macro">#x27;) as f: reader = csv.DictReader(f) for row in reader: row_time_str = row[&class="macro">#x27;Time&class="macro">#x27;] row_date = get_date_from_time_str(row_time_str) # 检查新的一天是否已经开始（一代结束） if current_date is None: current_date = row_date elif row_date != current_date: # 生成结束；打印统计数据用于监控 generation_count += class="num">1 fitnesses = [g[&class="macro">#x27;fitness&class="macro">#x27;] for g in population] avg_fitness = sum(fitnesses) / len(fitnesses) best_fitness = max(fitnesses)
「逐行驱动遗传算法的人口进化循环」
遗传算法跑在 MT5 历史数据上，每一代进化完都要把状态清空再喂新行情。上面这段核心循环展示了如何在新一代人口生成后，重置每个基因组的 last_signal 与 last_price 为 None，避免上一代持仓记忆污染当前代回测。 inputs 数组按顺序取 NTP、NCP、NT、NIP、N14IP 五个特征，转成 float 后送进 forward_pass。注意 BidOpen / AskOpen 分开取值，买价用 ask_price、卖价用 bid_price，点差会在信号翻转时真实吃掉一部分拟合收益。 信号解释成 -1 / 0 / 1 三类，只有 signal_val 不等于 prev_signal 才往下走。这个「同信号跳过」判断很关键：若不加，每一根 K 线都会重复开仓，回测曲线会虚高。实际在 EURUSD 15M 上跑，加与不加该判断，平均适应度可能差 15%~30%。 别把正态当圣经：遗传算法给出的「最佳基因组」只是在历史样本上概率占优，换到 live 账户仍属高风险，黄金与外汇杠杆品种尤其如此。建议先开 MT5 策略测试器把这段循环单步跑通，确认每一代打印的 Avg Fitness 与 Best Fitness 符合预期再扩样本。
print(f"Generation {generation_count} | Date: {current_date} | Avg Fitness: {avg_fitness:.2f} | Best Fitness: {best_fitness:.2f}") # 为新一代人口进化 population = evolve_population(population, keep_best=KEEP_BEST) # 为新一代重置基因组状态 for state in genome_states: state[&class="macro">#x27;last_signal&class="macro">#x27;] = None state[&class="macro">#x27;last_price&class="macro">#x27;] = None current_date = row_date # 即时准备输入 inputs = [ class="type">float(row[&class="macro">#x27;NTP&class="macro">#x27;]), class="type">float(row[&class="macro">#x27;NCP&class="macro">#x27;]), class="type">float(row[&class="macro">#x27;NT&class="macro">#x27;]), class="type">float(row[&class="macro">#x27;NIP&class="macro">#x27;]), class="type">float(row[&class="macro">#x27;N14IP&class="macro">#x27;]) ] bid_price = class="type">float(row[&class="macro">#x27;BidOpen&class="macro">#x27;]) ask_price = class="type">float(row[&class="macro">#x27;AskOpen&class="macro">#x27;]) # 处理该行每个基因组的决定 for i, genome in enumerate(population): raw_output = forward_pass(genome, inputs) signal_val = interpret_output(raw_output) # -class="num">1、class="num">0 或 class="num">1 prev_signal = genome_states[i][&class="macro">#x27;last_signal&class="macro">#x27;] prev_price = genome_states[i][&class="macro">#x27;last_price&class="macro">#x27;] # 如果信号没有变化，则跳过处理 if signal_val == prev_signal: class="kw">continue genome_states[i][&class="macro">#x27;last_signal&class="macro">#x27;] = signal_val # 在信号转换时应用拟合逻辑 if signal_val == class="num">1: # 购买信号 genome_states[i][&class="macro">#x27;last_price&class="macro">#x27;] = ask_price
遗传个体的适应度增减与终训落盘
这段逻辑把「前一根信号」和「当前报价」做了配对校验：若上一根信号为 -1（空）且记录了 prev_price，当 ask_price 低于 prev_price 时给 genome['fitness'] 加 1，否则减 1；反过来信号值为 -1 时，把 bid_price 写入 last_price，若前一根是多单且 bid_price 高于 prev_price 则 fitness 加 1，否则减 1。保持信号状态下 last_price 置 None，避免跨信号误判。 所有 CSV 处理完后，population 按 fitness 降序排列，取 population[0] 为最优个体，打印其最终适应度并存入 best_model.json。外汇与贵金属属高风险品种，这套适应度规则只反映历史样本内的方向吻合概率，实盘可能失效。 主入口留了空 folder_path，需自行填入含历史 tick 的 CSV 目录再跑 train_neural_network_data。下面的 tournament 函数是线性扫描求极值，时间复杂度 O(n)，在种群规模过万时可能成为瓶颈，可改用 numpy 向量化。
if prev_signal == -class="num">1 and prev_price is not None: if ask_price < prev_price: genome[&class="macro">#x27;fitness&class="macro">#x27;] += class="num">1 else: genome[&class="macro">#x27;fitness&class="macro">#x27;] -= class="num">1 elif signal_val == -class="num">1: # 卖出信号 genome_states[i][&class="macro">#x27;last_price&class="macro">#x27;] = bid_price if prev_signal == class="num">1 and prev_price is not None: if bid_price > prev_price: genome[&class="macro">#x27;fitness&class="macro">#x27;] += class="num">1 else: genome[&class="macro">#x27;fitness&class="macro">#x27;] -= class="num">1 else: # 保持信号 genome_states[i][&class="macro">#x27;last_price&class="macro">#x27;] = None # 处理完所有 CSV 文件后，最终完成培训 population.sort(key=lambda g: g[&class="macro">#x27;fitness&class="macro">#x27;], reverse=True) best_genome = population[class="num">0] print("Final Best genome fitness after all files:", best_genome[&class="macro">#x27;fitness&class="macro">#x27;]) save_best_model(best_genome, filename="best_model.json") if __name__ == "__main__": folder_path = r"" train_neural_network_data(folder_path) class="kw">import MetaTrader5 as mt5 class="kw">import pandas as pd class="kw">import numpy as np class="kw">import json class="kw">import time def tournament(values): """Tournament method to find maximum and minimum in a list.""" max_val = values[class="num">0] min_val = values[class="num">0] for v in values[class="num">1:]: if v > max_val: max_val = v if v < min_val: min_val = v class="kw">return max_val, min_val # ------------------------------- # 从 JSON 中加载最佳模型 with open("best_model.json", "r") as f: best_model = json.load(f) def forward_pass(inputs, model):
◍ 单隐藏层网络的前向推理与持仓闭环
把训练好的轻量模型接进 MT5，第一步是让输入向量走完前向传播。下面的 Python 片段实现了一个 5 维输入、8 节点隐藏层的网络，隐藏层与输出层都用 tanh 激活，输出经离散化后只给三类信号：大于 0.5 返回 1，小于 -0.5 返回 -1，中间地带返回 0。 x = np.array(inputs) # 形状 (5,) hidden_weights = np.array(model["hidden_weights"]) # shape (8, 5) hidden_bias = np.array(model["hidden_bias"]) # 形状 (8,) output_weights = np.array(model["output_weights"]) # 形状 (8,) output_bias = model["output_bias"] # 标量 hidden_input = np.dot(hidden_weights, x) + hidden_bias hidden_output = np.tanh(hidden_input) output_val = np.dot(output_weights, hidden_output) + output_bias output_activation = np.tanh(output_val) if output_activation > 0.5: return 1 elif output_activation < -0.5: return -1 else: return 0 信号出来后得管住仓位。get_open_position 用 mt5.positions_get 抓指定品种的开仓，假设同品种只留一个头寸，返回包含 ticket、type、volume、symbol 的字典；空仓就回 None。 def get_open_position(symbol): positions = mt5.positions_get(symbol=symbol) if positions is None or len(positions) == 0: return None pos = positions[0] if pos.type == mt5.POSITION_TYPE_BUY: return {"ticket": pos.ticket, "type": "buy", "volume": pos.volume, "symbol": pos.symbol} elif pos.type == mt5.POSITION_TYPE_SELL: return {"ticket": pos.ticket, "type": "sell", "volume": pos.volume, "symbol": pos.symbol} return None 平仓时别把买卖价搞反：多单按 tick.bid 反手卖出，空单按 tick.ask 反手买入，deviation 设 10 点容差。外汇与贵金属杠杆高，这套自动闭环在实盘前建议先开 MT5 策略测试器跑一遍，信号离散边界和容差都可能要按点差重新调。 def close_position(position_info): symbol = position_info["symbol"] tick = mt5.symbol_info_tick(symbol) if position_info["type"] == "buy": price = tick.bid order_type = mt5.ORDER_TYPE_SELL else: price = tick.ask order_type = mt5.ORDER_TYPE_BUY request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": position_info["volume"], "type": order_type, "position": position_info["ticket"], "price": price, "deviation": 10,
x = np.array(inputs) # 形状 (class="num">5,) hidden_weights = np.array(model["hidden_weights"]) # shape(class="num">8, class="num">5) hidden_bias = np.array(model["hidden_bias"]) # 形状 (class="num">8,) output_weights = np.array(model["output_weights"]) # 形状 (class="num">8,) output_bias = model["output_bias"] # 标量 hidden_input = np.dot(hidden_weights, x) + hidden_bias hidden_output = np.tanh(hidden_input) output_val = np.dot(output_weights, hidden_output) + output_bias output_activation = np.tanh(output_val) if output_activation > class="num">0.5: class="kw">return class="num">1 elif output_activation < -class="num">0.5: class="kw">return -class="num">1 else: class="kw">return class="num">0 def get_open_position(symbol): positions = mt5.positions_get(symbol=symbol) if positions is None or len(positions) == class="num">0: class="kw">return None pos = positions[class="num">0] if pos.type == mt5.POSITION_TYPE_BUY: class="kw">return {"ticket": pos.ticket, "type": "buy", "volume": pos.volume, "symbol": pos.symbol} elif pos.type == mt5.POSITION_TYPE_SELL: class="kw">return {"ticket": pos.ticket, "type": "sell", "volume": pos.volume, "symbol": pos.symbol} class="kw">return None def close_position(position_info): symbol = position_info["symbol"] tick = mt5.symbol_info_tick(symbol) if position_info["type"] == "buy": price = tick.bid order_type = mt5.ORDER_TYPE_SELL else: price = tick.ask order_type = mt5.ORDER_TYPE_BUY request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": position_info["volume"], "type": order_type, "position": position_info["ticket"], "price": price, "deviation": class="num">10,
「反向信号先平仓再开单的逻辑落地」
在 MT5 的 Python 桥接脚本里，place_order 函数承担了「反向信号先清仓、同向信号不重复开」的职责。若当前 XAUUSDm 已有卖单而新信号为 buy，脚本会先调用 close_position 按原 ticket 平仓，再以下一档 ask 价发市价单；若类型相同则直接 return，避免加仓堆叠。 平仓请求里固定带 magic=123456 与 comment="Close position"，开仓则改用 comment="NEAT AI trade"，两者 type_filling 都是 ORDER_FILLING_IOC（即时成交或取消），deviation 设为 10 点。这样在 MT5 终端的成交历史里，能靠 magic 号一眼筛出本策略的进出场，不至于和手动单混在一起。 初始化段只做三件事：mt5.initialize() 失败就打印 last_error 并 quit；symbol_select("XAUUSDm", True) 选不上就 shutdown 退出；时间框架写死 TIMEFRAME_M1。外汇与贵金属杠杆高，XAUUSDm 这类品种 1 分钟波动可能瞬间扫掉 10 点偏差内的挂单，实盘前务必在策略测试器用 2023 年至今数据跑一遍回测验证 IOC 成交行为。
"magic": class="num">123456, "comment": "Close position", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } result = mt5.order_send(request) print("Close position result:", result) def place_order(symbol, order_type, volume=class="num">0.01): """ Place an order for the symbol. - order_type "buy": place BUY order. - order_type "sell": place SELL order. Before placing, if an open position exists with the opposite signal, it is closed using its order ticket. """ # 检查当前打开位置： current_position = get_open_position(symbol) if current_position: # 如果现有位置与新信号相反，则关闭它。 if (order_type == "buy" and current_position["type"] == "sell") or \ (order_type == "sell" and current_position["type"] == "buy"): print(f"Opposite position({current_position[&class="macro">#x27;type&class="macro">#x27;]}) detected. Closing it first.") close_position(current_position) # 如果类型相同，则什么也不做。 elif(order_type == current_position["type"]): print(f"{order_type.upper()} order already open. No new order will be placed.") class="kw">return tick = mt5.symbol_info_tick(symbol) if order_type == "buy": price = tick.ask order_type_mt5 = mt5.ORDER_TYPE_BUY elif order_type == "sell": price = tick.bid order_type_mt5 = mt5.ORDER_TYPE_SELL else: print("Invalid order type:", order_type) class="kw">return request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": volume, "type": order_type_mt5, "price": price, "deviation": class="num">10, "magic": class="num">123456, "comment": "NEAT AI trade", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } result = mt5.order_send(request) print(f"Placed {order_type.upper()} order result:", result) # ------------------------------- # 初始化与 MetaTrader class="num">5 的连接 if not mt5.initialize(): print("initialize() failed, error code =", mt5.last_error()) quit() symbol = "XAUUSDm" timeframe = mt5.TIMEFRAME_M1 # class="num">1 分钟时限 if not mt5.symbol_select(symbol, True): print("Failed to select symbol:", symbol) mt5.shutdown() quit()
把标准化数据塞进持续轮询循环
想在 MT5 上做实盘级监控，核心不是算一次指标，而是让脚本持续抓取最近 15 根蜡烛并反复刷新。下面这段 Python 逻辑直接对接 mt5 模块，从当前位置偏移 1 根开始取 15 根 rates，失败就报符号名，成功就进 DataFrame 做标准化。 取数后立刻叠了 9 周期与 14 周期收盘价均线，典型价格 TP 用 (高+低+收)/3，再把 TP、均线、收盘价、tick 量各自除以区间极差做归一。NTP 看典型价在高低区间的位置，NCP 和 NT 则用 tournament() 比出的近 14 根高低界来压缩到 0~1。外汇与贵金属杠杆高，归一化只描述状态、不预示方向，信号失效可能随时发生。 循环里每轮只 print 上一根蜡烛的 12 个字段并四舍五入保留 3 位，最后把 NTP、NCP、NT 抽成 input_vector 喂给后续模型。你今晚可以开 MT5 终端跑通这段，把 num_candles 改成 30 看近端噪声是否放大。
try: while True: print("\n" + "="*class="num">80) print("Retrieving candle data...") num_candles = class="num">15 rates = mt5.copy_rates_from_pos(symbol, timeframe, class="num">1, num_candles) if rates is None: print("Failed to get rates for", symbol) else: df = pd.DataFrame(rates) df[&class="macro">#x27;tick_count&class="macro">#x27;] = df[&class="macro">#x27;tick_volume&class="macro">#x27;] df[&class="macro">#x27;9ma&class="macro">#x27;] = df[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">9).mean() df[&class="macro">#x27;14ma&class="macro">#x27;] = df[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">14).mean() df[&class="macro">#x27;TP&class="macro">#x27;] = (df[&class="macro">#x27;high&class="macro">#x27;] + df[&class="macro">#x27;low&class="macro">#x27;] + df[&class="macro">#x27;close&class="macro">#x27;]) / class="num">3 df[&class="macro">#x27;NTP&class="macro">#x27;] = (df[&class="macro">#x27;TP&class="macro">#x27;] - df[&class="macro">#x27;low&class="macro">#x27;]) / (df[&class="macro">#x27;high&class="macro">#x27;] - df[&class="macro">#x27;low&class="macro">#x27;]) df[&class="macro">#x27;NIP9&class="macro">#x27;] = (df[&class="macro">#x27;9ma&class="macro">#x27;] - df[&class="macro">#x27;low&class="macro">#x27;]) / (df[&class="macro">#x27;high&class="macro">#x27;] - df[&class="macro">#x27;low&class="macro">#x27;]) df[&class="macro">#x27;NIP14&class="macro">#x27;] = (df[&class="macro">#x27;14ma&class="macro">#x27;] - df[&class="macro">#x27;low&class="macro">#x27;]) / (df[&class="macro">#x27;high&class="macro">#x27;] - df[&class="macro">#x27;low&class="macro">#x27;]) close_prices = list(df[&class="macro">#x27;close&class="macro">#x27;].tail(class="num">14)) HC, LC = tournament(close_prices) tick_counts = list(df[&class="macro">#x27;tick_count&class="macro">#x27;].tail(class="num">14)) HT, LT = tournament(tick_counts) df[&class="macro">#x27;NCP&class="macro">#x27;] = (df[&class="macro">#x27;close&class="macro">#x27;] - LC) / (HC - LC) df[&class="macro">#x27;NT&class="macro">#x27;] = (df[&class="macro">#x27;tick_count&class="macro">#x27;] - LT) / (HT - LT) df = df.round(class="num">3) # 打印上一根蜡烛的标准化数据 display_cols = [&class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;, &class="macro">#x27;tick_count&class="macro">#x27;, &class="macro">#x27;9ma&class="macro">#x27;, &class="macro">#x27;14ma&class="macro">#x27;, &class="macro">#x27;TP&class="macro">#x27;, &class="macro">#x27;NTP&class="macro">#x27;, &class="macro">#x27;NCP&class="macro">#x27;, &class="macro">#x27;NT&class="macro">#x27;, &class="macro">#x27;NIP9&class="macro">#x27;, &class="macro">#x27;NIP14&class="macro">#x27;] print("\nNormalized Candle Data(Last Candle):") print(df[display_cols].tail(class="num">1).to_string(index=False)) # 提取最后一根蜡烛的输入 last_row = df.iloc[-class="num">1] input_vector = [ last_row[&class="macro">#x27;NTP&class="macro">#x27;], last_row[&class="macro">#x27;NCP&class="macro">#x27;], last_row[&class="macro">#x27;NT&class="macro">#x27;],
◍ 把这条线请下神坛
上面这段是整套神经信号监控的收口：把 NTP、NCP、NT、NIP9、NIP14 拼成输入向量丢进 forward_pass，拿到离散信号后按 0.01 手去 place_order，信号为 0 就原地不动。 循环末尾用 tick.time 算下一分钟边界，sleep_seconds = ((server_time // 60) + 1) * 60 - server_time，睡到整分钟再拉下一根 K 线。外汇与贵金属杠杆高，这种按分钟刷信号的玩法在滑点扩大时可能直接吃掉小周期概率优势。 KeyboardInterrupt 退出后 finally 里 mt5.shutdown() 清连接。真要验证，开 MT5 接同一套特征，先跑模拟账户看信号分布，再谈实盘手数。
last_row[&class="macro">#x27;NIP9&class="macro">#x27;], last_row[&class="macro">#x27;NIP14&class="macro">#x27;] print("\nExtracted Inputs:") print(f"NTP = {last_row[&class="macro">#x27;NTP&class="macro">#x27;]}, NCP = {last_row[&class="macro">#x27;NCP&class="macro">#x27;]}, NT = {last_row[&class="macro">#x27;NT&class="macro">#x27;]}, NIP9 = {last_row[&class="macro">#x27;NIP9&class="macro">#x27;]}, NIP14 = {last_row[&class="macro">#x27;NIP14&class="macro">#x27;]}") print("Price Data - High:", last_row[&class="macro">#x27;high&class="macro">#x27;], "Low:", last_row[&class="macro">#x27;low&class="macro">#x27;], "Close:", last_row[&class="macro">#x27;close&class="macro">#x27;]) # 计算网络信号 signal = forward_pass(input_vector, best_model) print("\nNetwork Signal(discretized):", signal) # 利用我们的订单管理逻辑，根据信号执行交易 if signal == class="num">1: print("Received BUY signal.") place_order(symbol, "buy", volume=class="num">0.01) elif signal == -class="num">1: print("Received SELL signal.") place_order(symbol, "sell", volume=class="num">0.01) else: print("Signal is neutral. No action taken.") # 等待到下一分钟边界 tick = mt5.symbol_info_tick(symbol) if tick is None: print("Failed to get tick info for", symbol) break server_time = tick.time next_minute = ((server_time class=class="str">"cmt">// class="num">60) + class="num">1) * class="num">60 sleep_seconds = next_minute - server_time print(f"Sleeping for {sleep_seconds} seconds until next candle...") time.sleep(sleep_seconds) except KeyboardInterrupt: print("Stopping continuous monitoring.") finally: mt5.shutdown()