文章 "神经网络实践：第一个神经元"
是的，没错，我刚在 YouTube 上看到一个视频，用 N.E.A.T 方法训练人工智能玩 flappy bird 和其他游戏，我就萌生了用 N.E.A.T 训练人工智能进行交易的想法，我刚刚学习了神经网络 的基础知识，并在 chat GPT 的帮助下创建了一个模型，因为我不会编程。我花了两周时间对数据进行归一化处理，花了两天时间创建模型，但只用了一个小时就完成了训练，1300 代，每代 20 个基因组，我那台 5 年前的二手笔记本电脑都快着火了，当我将模型连接到 MT5 时，模型非常激进，预测下一根蜡烛非常准确，但却无法盈利。但学习和看到模型预测很有趣，这也是我来这里学习更多人工智能知识的原因，这就是 NEAT 人工智能的代码。
import os import csv import math import random import json from datetime import datetime # ----------------------------------------- # UTILS：激活和简单前传 # ----------------------------------------- def tanh(x): return math.tanh(x) def forward_pass(genome, inputs): """ genome = { 'hidden_weights': 2D list, 'hidden_bias': 1D list, 'output_weights': list, 'output_bias': float, 'fitness': float } inputs: list of 5 floats Returns: single float in (-1, 1) """ hidden_activations = [] for h in range(len(genome['hidden_bias'])): z = sum(inputs[i] * genome['hidden_weights'][h][i] for i in range(len(inputs))) z += genome['hidden_bias'][h] hidden_activations.append(tanh(z)) z_out = sum(hidden_activations[h] * genome['output_weights'][h] for h in range(len(hidden_activations))) z_out += genome['output_bias'] return tanh(z_out) def interpret_output(output): """ Convert the Tanh output to discrete values: if output >= 0.5 => 1 (Buy) if output <= -0.5 => -1 (Sell) otherwise => 0 (Hold) """ if output >= 0.5: return 1 elif output <= -0.5: return -1 else: return 0 # ------------------------------------------------------ # 基因算法：初始化、选择、突变 # ------------------------------------------------------ def create_random_genome(input_size=5, hidden_size=8): return { 'hidden_weights': [[random.uniform(-1, 1) for _ in range(input_size)] for _ in range(hidden_size)], 'hidden_bias': [random.uniform(-1, 1) for _ in range(hidden_size)], 'output_weights': [random.uniform(-1, 1) for _ in range(hidden_size)], 'output_bias': random.uniform(-1, 1), 'fitness': 0.0 } def mutate(genome, mutation_rate=0.1, mutation_strength=0.5): for h in range(len(genome['hidden_weights'])): for i in range(len(genome['hidden_weights'][h])): if random.random() < mutation_rate: genome['hidden_weights'][h][i] += random.uniform(-mutation_strength, mutation_strength) for h in range(len(genome['hidden_bias'])): if random.random() < mutation_rate: genome['hidden_bias'][h] += random.uniform(-mutation_strength, mutation_strength) for h in range(len(genome['output_weights'])): if random.random() < mutation_rate: genome['output_weights'][h] += random.uniform(-mutation_strength, mutation_strength) if random.random() < mutation_rate: genome['output_bias'] += random.uniform(-mutation_strength, mutation_strength) def crossover(genome1, genome2): child = create_random_genome() # 随机初始值，将被覆盖 for h in range(len(genome1['hidden_weights'])): for i in range(len(genome1['hidden_weights'][h])): child['hidden_weights'][h][i] = (genome1['hidden_weights'][h][i] if random.random() < 0.5 else genome2['hidden_weights'][h][i]) for h in range(len(genome1['hidden_bias'])): child['hidden_bias'][h] = (genome1['hidden_bias'][h] if random.random() < 0.5 else genome2['hidden_bias'][h]) for h in range(len(genome1['output_weights'])): child['output_weights'][h] = (genome1['output_weights'][h] if random.random() < 0.5 else genome2['output_weights'][h]) child['output_bias'] = (genome1['output_bias'] if random.random() < 0.5 else genome2['output_bias']) child['fitness'] = 0.0 return child def evolve_population(population, keep_best=5): population.sort(key=lambda g: g['fitness'], reverse=True) new_population = population[:keep_best] # 保留最好的基因组 while len(new_population) < len(population): parent1 = random.choice(new_population) parent2 = random.choice(new_population) child = crossover(parent1, parent2) mutate(child) new_population.append(child) return new_population # ------------------------------------------- # 保存并加载最佳模型，避免重新训练 # ------------------------------------------- def save_best_model(genome, filename="best_model.json"): data = { 'hidden_weights': genome['hidden_weights'], 'hidden_bias': genome['hidden_bias'], 'output_weights': genome['output_weights'], 'output_bias': genome['output_bias'], 'fitness': genome['fitness'] } with open(filename, 'w') as f: json.dump(data, f, indent=2) def load_best_model(filename="best_model.json"): if not os.path.exists(filename): return None with open(filename, 'r') as f: data = json.load(f) return data # ---------------- # 主要训练逻辑 # ---------------- def train_neural_network_data(folder_path): # 超参数 POP_SIZE = 20 KEEP_BEST = 5 input_size = 5 hidden_size = 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[0]['hidden_weights'] = best_saved['hidden_weights'] population[0]['hidden_bias'] = best_saved['hidden_bias'] population[0]['output_weights'] = best_saved['output_weights'] population[0]['output_bias'] = best_saved['output_bias'] population[0]['fitness'] = 0.0 # 帮助解析包含时区信息的 ISO8601 日期/时间字符串 def get_date_from_time_str(t_str): dt = datetime.strptime(t_str, "%Y-%m-%dT%H:%M:%S%z") return dt.date() csv_files = sorted([f for f in os.listdir(folder_path) if f.endswith('.csv')]) # 对于每个基因组，存储最后一个信号和相应的价格（如果有的话） genome_states = [{'last_signal': None, 'last_price': None} for _ in range(POP_SIZE)] current_date = None generation_count = 0 # 一次处理一个 CSV 文件（逐行流式处理） for csv_file in csv_files: file_path = os.path.join(folder_path, csv_file) with open(file_path, 'r', newline='') as f: reader = csv.DictReader(f) for row in reader: row_time_str = row['Time'] 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 += 1 fitnesses = [g['fitness'] for g in population] avg_fitness = sum(fitnesses) / len(fitnesses) best_fitness = max(fitnesses) print(f"Generation {generation_count} | Date: {current_date} | Avg Fitness: {avg_fitness:.2 f} | Best Fitness: {best_fitness:.2 f}") # 为新一代人口进化 population = evolve_population(population, keep_best=KEEP_BEST) # 为新一代重置基因组状态 for state in genome_states: state['last_signal'] = None state['last_price'] = None current_date = row_date # 即时准备输入 inputs = [ float(row['NTP']), float(row['NCP']), float(row['NT']), float(row['NIP']), float(row['N14IP']) ] bid_price = float(row['BidOpen']) ask_price = float(row['AskOpen']) # 处理该行每个基因组的决定 for i, genome in enumerate(population): raw_output = forward_pass(genome, inputs) signal_val = interpret_output(raw_output) # -1、0 或 1 prev_signal = genome_states[i]['last_signal'] prev_price = genome_states[i]['last_price'] # 如果信号没有变化，则跳过处理 if signal_val == prev_signal: continue genome_states[i]['last_signal'] = signal_val # 在信号转换时应用拟合逻辑 if signal_val == 1: # 购买信号 genome_states[i]['last_price'] = ask_price if prev_signal == -1 and prev_price is not None: if ask_price < prev_price: genome['fitness'] += 1 else: genome['fitness'] -= 1 elif signal_val == -1: # 卖出信号 genome_states[i]['last_price'] = bid_price if prev_signal == 1 and prev_price is not None: if bid_price > prev_price: genome['fitness'] += 1 else: genome['fitness'] -= 1 else: # 保持信号 genome_states[i]['last_price'] = None # 处理完所有 CSV 文件后，最终完成培训 population.sort(key=lambda g: g['fitness'], reverse=True) best_genome = population[0] print("Final Best genome fitness after all files:", best_genome['fitness']) save_best_model(best_genome, filename="best_model.json") if __name__ == "__main__": folder_path = r"" train_neural_network_data(folder_path)和 MT5
import MetaTrader5 as mt5 import pandas as pd import numpy as np import json import time def tournament(values): """Tournament method to find maximum and minimum in a list.""" max_val = values[0] min_val = values[0] for v in values[1:]: if v > max_val: max_val = v if v < min_val: min_val = v return max_val, min_val # ------------------------------- # 从 JSON 中加载最佳模型 with open("best_model.json", "r") as f: best_model = json.load(f) def forward_pass(inputs, model): """ Perform a forward pass through a feedforward network with one hidden layer. Uses tanh for both hidden and output layers. The output is discretized: >0.5 --> 1, < -0.5 --> -1, else 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 # ------------------------------- # 利用职位实现订单管理功能 def get_open_position(symbol): """ Returns information about an open position for the symbol as a dictionary: {"ticket": ticket, "type": "buy"/"sell", "volume": volume}. If no position is open, returns None. """ 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 def close_position(position_info): """ Closes the given position using its ticket. """ 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, "magic": 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=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['type']}) 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.") 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) return request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": volume, "type": order_type_mt5, "price": price, "deviation": 10, "magic": 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 5 的连接 if not mt5.initialize(): print("initialize() failed, error code =", mt5.last_error()) quit() symbol = "XAUUSDm" timeframe = mt5.TIMEFRAME_M1 # 1 分钟时限 if not mt5.symbol_select(symbol, True): print("Failed to select symbol:", symbol) mt5.shutdown() quit() # 持续监测循环 try: while True: print("\n" + "="*80) print("Retrieving candle data...") num_candles = 15 rates = mt5.copy_rates_from_pos(symbol, timeframe, 1, num_candles) if rates is None: print("Failed to get rates for", symbol) else: df = pd.DataFrame(rates) df['tick_count'] = df['tick_volume'] df['9ma'] = df['close'].rolling(window=9).mean() df['14ma'] = df['close'].rolling(window=14).mean() df['TP'] = (df['high'] + df['low'] + df['close']) / 3 df['NTP'] = (df['TP'] - df['low']) / (df['high'] - df['low']) df['NIP9'] = (df['9ma'] - df['low']) / (df['high'] - df['low']) df['NIP14'] = (df['14ma'] - df['low']) / (df['high'] - df['low']) close_prices = list(df['close'].tail(14)) HC, LC = tournament(close_prices) tick_counts = list(df['tick_count'].tail(14)) HT, LT = tournament(tick_counts) df['NCP'] = (df['close'] - LC) / (HC - LC) df['NT'] = (df['tick_count'] - LT) / (HT - LT) df = df.round(3) # 打印上一根蜡烛的标准化数据 display_cols = ['high', 'low', 'close', 'tick_count', '9ma', '14ma', 'TP', 'NTP', 'NCP', 'NT', 'NIP9', 'NIP14'] print("\nNormalized Candle Data (Last Candle):") print(df[display_cols].tail(1).to_string(index=False)) # 提取最后一根蜡烛的输入 last_row = df.iloc[-1] input_vector = [ last_row['NTP'], last_row['NCP'], last_row['NT'], last_row['NIP9'], last_row['NIP14'] ] print("\nExtracted Inputs:") print(f"NTP = {last_row['NTP']}, NCP = {last_row['NCP']}, NT = {last_row['NT']}, NIP9 = {last_row['NIP9']}, NIP14 = {last_row['NIP14']}") print("Price Data - High:", last_row['high'], "Low:", last_row['low'], "Close:", last_row['close']) # 计算网络信号 signal = forward_pass(input_vector, best_model) print("\nNetwork Signal (discretized):", signal) # 利用我们的订单管理逻辑，根据信号执行交易 if signal == 1: print("Received BUY signal.") place_order(symbol, "buy", volume=0.01) elif signal == -1: print("Received SELL signal.") place_order(symbol, "sell", volume=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 // 60) + 1) * 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()
非常好
Multi Dead 神经网络 的基础知识，并在 GPT 聊天工具的帮助下创建了一个模型，因为我不知道如何编程。我花了两周时间将数据正常化，花了两天时间创建模型，但只用了一个小时就完成了训练，1300 代，每代 20 个基因组，我那台 5 年前的二手笔记本电脑都快着火了，当我将模型连接到 MT5 时，模型非常激进，预测下一根蜡烛非常准确，但并没有盈利。但学习和观察模型的预测很有趣，这也是我来这里学习更多人工智能知识的原因，这就是 NEAT 人工智能代码
神经网络是一个非常有趣和好玩的课题。不过，由于我决定先完成重放/模拟器，所以我暂时不做解释。不过，一旦我完成了关于模拟器的文章发布，我们就会重新发布关于神经网络的新文章。我们的目标始终是展示它们在引擎盖下是如何工作的。大多数人认为神经网络是神奇的代码，其实不然。但它们仍然是一个有趣的主题。我甚至在考虑建立一些模型，让每个人都能看到神经网络是如何在没有监督或先验数据的情况下学习的。这非常有趣，有助于理解某些事情。详情：我的所有代码都将用 MQL5 编写。既然你说你不是程序员。学习 MQL5 并开始实施自己的解决方案如何？我正在写一系列针对您这样的人的文章。最新的一篇可以在这里看到：https://www.mql5.com/zh/articles/15833。 在这一系列文章中，我从最基础的东西开始讲解。因此，如果你对编程一无所知，请回到本系列的第一篇文章。文章开头都有指向前几篇文章的链接。
Do básico ao intermediário: Indicador (IV)
- www.mql5.com
Neste artigo, vermos como é fácil de criar e implementar uma metodologia operacional, visando colorir candles. Sendo este um conceito, que diversos operadores apreciam imensamente. Porém, é preciso se tomar cuidado ao implementar tal tipo de coisa. Isto para que as barras, ou candles, mantenham a sua aparência original. Visando assim não prejudicar a leitura que muitos operadores fazem candle a candle.
非常感谢你的介绍！
新文章 神经网络实践：第一个神经元已发布：
如果您仔细阅读以前的文章，您会注意到我们必须采用某些数学方法来确定表示斜率的常数 <a> 和表示截距的常数 <b> 的最佳值。这些调整使我们能够找到最合适的线性方程。我们探索了两种方法来实现这一点：一种通过导数计算，另一种通过矩阵计算。
然而，从现在开始，这样的计算将不再有用。这是因为我们需要设计一种替代方法来确定线性方程中的常数。在上一篇文章中，我演示了如何找到表示斜率的常数。我希望您喜欢尝试该代码，因为现在我们将要处理稍微复杂一些的问题。然而，尽管挑战性只增加了一点点，但下一步将释放无数的可能性。事实上，这可能是我们神经网络系列中最有趣的文章，因为接下来的一切都将变得更加简单和实用。
作者：Daniel Jose