137 lignes
6,4 Kio
Python
137 lignes
6,4 Kio
Python
# Copyright 2026, Allan Munene Mutiiria.
| |||
# https://t.me/Forex_Algo_Trader
| |||
| |||
import os
| |||
import numpy as np
| |||
import onnx
| |||
from onnx import helper, TensorProto, numpy_helper
| |||
| |||
# Numbers going in
| |||
INPUTS = 16
| |||
# Width of each branch before they rejoin
| |||
BRANCH = 8
| |||
# Width after the two branches are concatenated
| |||
MERGED = 10
| |||
# Down, flat, up
| |||
OUTPUTS = 3
| |||
| |||
rng = np.random.default_rng(7)
| |||
| |||
def xavier(rows, cols):
| |||
# Keeps the signal from dying or exploding
| |||
bound = np.sqrt(6.0 / (rows + cols))
| |||
return rng.uniform(-bound, bound, (rows, cols)).astype(np.float32)
| |||
| |||
# Stored through float_data rather than raw_data, so both paths are covered
| |||
Wa = xavier(INPUTS, BRANCH)
| |||
initializers = [
| |||
helper.make_tensor("Wa", TensorProto.FLOAT, [INPUTS, BRANCH],
| |||
Wa.flatten().tolist(), raw=False),
| |||
numpy_helper.from_array(np.zeros(BRANCH, np.float32), "ba"),
| |||
numpy_helper.from_array(xavier(INPUTS, BRANCH), "Wb"),
| |||
numpy_helper.from_array(np.zeros(BRANCH, np.float32), "bb"),
| |||
numpy_helper.from_array(xavier(BRANCH * 2, MERGED), "Wm"),
| |||
numpy_helper.from_array(np.zeros(MERGED, np.float32), "bm"),
| |||
numpy_helper.from_array(xavier(MERGED, OUTPUTS), "Wo"),
| |||
numpy_helper.from_array(np.zeros(OUTPUTS, np.float32), "bo"),
| |||
]
| |||
| |||
# Stored as double so the parser meets a type other than float
| |||
initializers.append(numpy_helper.from_array(np.full(MERGED, 0.85, np.float64), "scale"))
| |||
| |||
# Stored through int64_data, which is a third way a tensor can carry values
| |||
initializers.append(helper.make_tensor("reshape_to", TensorProto.INT64, [2],
| |||
[1, BRANCH * 2], raw=False))
| |||
| |||
nodes = [
| |||
# The input feeds two branches at once, so the graph is not a chain
| |||
helper.make_node("Gemm", ["input", "Wa", "ba"], ["gemm_a"], name="branch_a"),
| |||
helper.make_node("Tanh", ["gemm_a"], ["act_a"], name="tanh_a"),
| |||
helper.make_node("Gemm", ["input", "Wb", "bb"], ["gemm_b"], name="branch_b"),
| |||
helper.make_node("Relu", ["gemm_b"], ["act_b"], name="relu_b"),
| |||
# Two edges arrive here, so a layout cannot assume one parent
| |||
helper.make_node("Concat", ["act_a", "act_b"], ["merged"], axis=1, name="join"),
| |||
# Carries a list attribute, which is a form a scalar never exercises
| |||
helper.make_node("Transpose", ["merged"], ["flipped"], perm=[1, 0], name="flip"),
| |||
# Takes its shape from a tensor rather than an attribute
| |||
helper.make_node("Reshape", ["flipped", "reshape_to"], ["restored"], name="restore"),
| |||
helper.make_node("Gemm", ["restored", "Wm", "bm"], ["gemm_m"], name="mixer"),
| |||
helper.make_node("Sigmoid", ["gemm_m"], ["act_m"], name="sigmoid_m"),
| |||
# A node whose second input is an initializer rather than another node
| |||
helper.make_node("Mul", ["act_m", "scale_f"], ["scaled"], name="rescale"),
| |||
helper.make_node("Gemm", ["scaled", "Wo", "bo"], ["gemm_o"], name="output_layer"),
| |||
helper.make_node("Softmax", ["gemm_o"], ["output"], axis=1, name="softmax",
| |||
doc_string="Turns the three scores into probabilities"),
| |||
]
| |||
| |||
# The double tensor has to be cast before Mul can use it
| |||
nodes.insert(9, helper.make_node("Cast", ["scale"], ["scale_f"],
| |||
to=TensorProto.FLOAT, name="cast_scale"))
| |||
| |||
# Holds text rather than numbers, which the reader stores a different way
| |||
initializers.append(helper.make_tensor("labels", TensorProto.STRING, [OUTPUTS],
| |||
[b"down", b"flat", b"up"], raw=False))
| |||
| |||
# Named only where a value passes between layers, so every edge can be labelled
| |||
inner = [
| |||
helper.make_tensor_value_info("gemm_a", TensorProto.FLOAT, [1, BRANCH]),
| |||
helper.make_tensor_value_info("act_a", TensorProto.FLOAT, [1, BRANCH]),
| |||
helper.make_tensor_value_info("gemm_b", TensorProto.FLOAT, [1, BRANCH]),
| |||
helper.make_tensor_value_info("act_b", TensorProto.FLOAT, [1, BRANCH]),
| |||
helper.make_tensor_value_info("merged", TensorProto.FLOAT, [1, BRANCH * 2]),
| |||
helper.make_tensor_value_info("flipped", TensorProto.FLOAT, [BRANCH * 2, 1]),
| |||
helper.make_tensor_value_info("restored", TensorProto.FLOAT, [1, BRANCH * 2]),
| |||
helper.make_tensor_value_info("gemm_m", TensorProto.FLOAT, [1, MERGED]),
| |||
helper.make_tensor_value_info("act_m", TensorProto.FLOAT, [1, MERGED]),
| |||
helper.make_tensor_value_info("scale_f", TensorProto.FLOAT, [MERGED]),
| |||
helper.make_tensor_value_info("scaled", TensorProto.FLOAT, [1, MERGED]),
| |||
helper.make_tensor_value_info("gemm_o", TensorProto.FLOAT, [1, OUTPUTS]),
| |||
]
| |||
| |||
# Stored as an index and value pair rather than a dense run
| |||
sparse = helper.make_sparse_tensor(
| |||
helper.make_tensor("sp_values", TensorProto.FLOAT, [3], [0.5, 1.5, 2.5], raw=False),
| |||
helper.make_tensor("sp_index", TensorProto.INT64, [3], [0, 4, 9], raw=False),
| |||
[MERGED])
| |||
| |||
graph = helper.make_graph(
| |||
nodes, "viewer_sample",
| |||
[helper.make_tensor_value_info("input", TensorProto.FLOAT, [1, INPUTS])],
| |||
[helper.make_tensor_value_info("output", TensorProto.FLOAT, [1, OUTPUTS])],
| |||
initializer=initializers,
| |||
value_info=inner,
| |||
sparse_initializer=[sparse],
| |||
doc_string="Sample graph carrying every structure the reader handles",
| |||
)
| |||
| |||
model = helper.make_model(graph, producer_name="onnx-model-viewer",
| |||
opset_imports=[helper.make_opsetid("", 13)])
| |||
model.ir_version = 8
| |||
model.producer_version = "1.0"
| |||
model.domain = "forex.algo.trader"
| |||
model.model_version = 1
| |||
model.doc_string = "A viewer sample built to exercise every field"
| |||
| |||
# Key and value pairs the reader reports under metadata
| |||
for key, value in (("author", "Allan Munene Mutiiria"),
| |||
("purpose", "ONNX Model Viewer sample"),
| |||
("built", "make_model.py")):
| |||
entry = model.metadata_props.add()
| |||
entry.key = key
| |||
entry.value = value
| |||
onnx.checker.check_model(model)
| |||
| |||
# Climb to the terminal's MQL5 root from wherever this script was placed
| |||
folder = os.path.dirname(os.path.abspath(__file__))
| |||
while os.path.basename(folder) != "MQL5" and os.path.dirname(folder) != folder:
| |||
folder = os.path.dirname(folder)
| |||
| |||
# MQL5 can only read from Files, so the model has to land there
| |||
target = os.path.join(folder, "Files", "ONNX Model Viewer")
| |||
os.makedirs(target, exist_ok=True)
| |||
path = os.path.join(target, "model.onnx")
| |||
onnx.save(model, path)
| |||
| |||
params = sum(int(np.prod(t.dims)) for t in initializers)
| |||
print(f"wrote {path}")
| |||
print(f"{len(nodes)} nodes, {params} parameters")
|