138 lines
No EOL
6.4 KiB
Python
138 lines
No EOL
6.4 KiB
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 Part 1")
|
|
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") |