From 07e18043145f3d91c8ea7ec3ad41b3c905585277 Mon Sep 17 00:00:00 2001 From: Ethan Werner Date: Wed, 1 Jul 2026 09:42:39 +0400 Subject: initial commit: ts time series analysis library v0.1.0 - 354 financial and mathematical time-series indicators - Uniform C11 API with consistent function signatures - 349 passing tests with -Werror - Static and shared library builds - pkg-config integration - Python bindings - Fuzzing and sanitizer support - Security hardening flags --- scripts/gen_all.py | 135 ++++++++++++++++++++++++ scripts/gen_bindings.py | 274 ++++++++++++++++++++++++++++++++++++++++++++++++ scripts/gen_flat.py | 96 +++++++++++++++++ scripts/gen_fuzz.py | 74 +++++++++++++ scripts/gen_registry.py | 103 ++++++++++++++++++ scripts/ts_parser.py | 159 ++++++++++++++++++++++++++++ 6 files changed, 841 insertions(+) create mode 100644 scripts/gen_all.py create mode 100644 scripts/gen_bindings.py create mode 100644 scripts/gen_flat.py create mode 100644 scripts/gen_fuzz.py create mode 100644 scripts/gen_registry.py create mode 100644 scripts/ts_parser.py (limited to 'scripts') diff --git a/scripts/gen_all.py b/scripts/gen_all.py new file mode 100644 index 0000000..dd9d395 --- /dev/null +++ b/scripts/gen_all.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +"""Generate ts_generated.h with all indicator declarations and test registry. + +Parses lib/*.c for ts_parameter_t declarations and emits: +- extern ts_parameter_t const ts__p +- uint_t ts__offset(...) +- int ts_(...) +- bool ts_test_(void) [inside #ifdef TS_IMPLEMENTATION] +""" + +import importlib.util +import os +import re +import sys + +_spec = importlib.util.spec_from_file_location( + "ts_parser", + os.path.join(os.path.dirname(os.path.abspath(__file__)), "ts_parser.py"), +) +if _spec and _spec.loader: + _ts_parser = importlib.util.module_from_spec(_spec) + _spec.loader.exec_module(_ts_parser) + parse_all = _ts_parser.parse_all +else: + print("Error: could not load ts_parser", file=sys.stderr) + sys.exit(1) + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +TEST_DIR = os.path.join(ROOT, "test") +OUT_H = os.path.join(ROOT, "ts_generated.h") +OUT_TEST = os.path.join(TEST_DIR, "main_generated.c") + + +def generate_header(funcs): + """Generate ts_generated.h.""" + lines = [] + lines.append("/* Auto-generated indicator declarations. Do not edit. */") + lines.append("") + lines.append("#ifndef TS_GENERATED_H") + lines.append("#define TS_GENERATED_H") + lines.append("") + + for fm in funcs: + lines.append("extern ts_parameter_t const ts_" + fm.name + "_p;") + lines.append("uint_t ts_" + fm.name + "_offset( ts_constant_t const * );") + lines.append( + "int ts_" + + fm.name + + "( fp_t * const *y, uint_t y_n, fp_t const * const *x, " + "uint_t x_n, uint_t n, ts_constant_t const *c );" + ) + + lines.append("") + lines.append("#ifdef TS_IMPLEMENTATION") + lines.append("") + for fm in funcs: + lines.append("bool ts_test_" + fm.name + "( void );") + lines.append("") + lines.append("#endif /* TS_IMPLEMENTATION */") + lines.append("") + lines.append("#endif /* TS_GENERATED_H */") + lines.append("") + + return "\n".join(lines) + + +def generate_test_main(funcs): + """Generate test/main_generated.c with extern declarations and test array.""" + lines = [] + lines.append("/* Auto-generated test registry. Do not edit. */") + lines.append("") + lines.append("#define EXTERN_TEST(name) extern bool ts_test_##name(void)") + for fm in funcs: + lines.append("EXTERN_TEST(" + fm.name + ");") + lines.append("#undef EXTERN_TEST") + lines.append("") + lines.append("typedef struct {") + lines.append(" const char *name;") + lines.append(" bool (*func)(void);") + lines.append("} test_case_t;") + lines.append("") + lines.append("#define TEST_ENTRY(name) { #name, ts_test_##name }") + lines.append("") + lines.append("static test_case_t tests[] = {") + for fm in funcs: + lines.append(" TEST_ENTRY(" + fm.name + "),") + lines.append("};") + lines.append("") + lines.append("#define NUM_TESTS (sizeof(tests) / sizeof(tests[0]))") + lines.append("") + lines.append("#undef TEST_ENTRY") + lines.append("") + + return "\n".join(lines) + + +def _is_valid_c_identifier(name): + return bool(re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", name)) + + +def main(): + funcs = parse_all(detect_x_access=False) + + if not funcs: + print("Warning: no ts_parameter declarations found.", file=sys.stderr) + return 1 + + for fm in funcs: + if not _is_valid_c_identifier(fm.name): + print(f"Error: invalid C identifier '{fm.name}'", file=sys.stderr) + return 1 + + os.makedirs(TEST_DIR, exist_ok=True) + + try: + with open(OUT_H, "w", encoding="utf-8") as f: + f.write(generate_header(funcs)) + except OSError as e: + print(f"Error writing {OUT_H}: {e}", file=sys.stderr) + return 1 + + try: + with open(OUT_TEST, "w", encoding="utf-8") as f: + f.write(generate_test_main(funcs)) + except OSError as e: + print(f"Error writing {OUT_TEST}: {e}", file=sys.stderr) + return 1 + + print(f"Generated {len(funcs)} declarations -> {OUT_H}") + print(f"Generated test registry -> {OUT_TEST}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/gen_bindings.py b/scripts/gen_bindings.py new file mode 100644 index 0000000..845c8a5 --- /dev/null +++ b/scripts/gen_bindings.py @@ -0,0 +1,274 @@ +#!/usr/bin/env python3 +"""Generate python/ts/__init__.py from C source. + +Produces a compact module with a single _dispatch() function and a +FUNC_META lookup table instead of 350+ individual wrappers. +""" + +import importlib.util +import os +import py_compile +import sys + +_spec = importlib.util.spec_from_file_location( + "ts_parser", + os.path.join(os.path.dirname(os.path.abspath(__file__)), "ts_parser.py"), +) +if _spec and _spec.loader: + _ts_parser = importlib.util.module_from_spec(_spec) + _spec.loader.exec_module(_ts_parser) + parse_all = _ts_parser.parse_all +else: + print("Error: could not load ts_parser", file=sys.stderr) + sys.exit(1) + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +OUT = os.path.join(ROOT, "python", "ts", "__init__.py") + +DEFAULT_PARAMS = { + "ichimoku": {"tenkan": 9, "kijun": 26, "senkou": 52}, +} + +MIN_PARAM_DEFAULTS = { + "hull_ma": {"k": 2}, + "variable_index_dynamic_average": {"period": 2}, + "logarithm": {"b": 2.0}, + "choppiness_index": {"k": 2}, + "performance_ratio": {"k": 2}, + "standard_error": {"k": 2}, +} + + +def _py_default(pytype, py_fname, name): + if py_fname in DEFAULT_PARAMS.get(name, {}): + return DEFAULT_PARAMS[name][py_fname] + if py_fname in MIN_PARAM_DEFAULTS.get(name, {}): + return MIN_PARAM_DEFAULTS[name][py_fname] + return 1 if "uint64" in pytype else 1.0 + + +def main(): + funcs = parse_all(detect_x_access=True) + if not funcs: + print("Warning: no ts_parameter declarations found.", file=sys.stderr) + sys.exit(1) + + os.makedirs(os.path.dirname(OUT), exist_ok=True) + + with open(OUT, "w", encoding="utf-8") as f: + # --- Header --- + f.write('"""Python bindings for ts (time series) C library.\n') + f.write("\nAuto-generated. Uses a single dispatch function backed by\n") + f.write('a FUNC_META lookup table.\n"""\n\n') + + f.write("import ctypes\n") + f.write("import ctypes.util\n") + f.write("import os\n") + f.write("import numpy as np\n") + f.write("import threading\n\n") + + f.write("_tls = threading.local()\n\n") + + # --- Library loader --- + f.write("def _find_lib():\n") + f.write(' """Try system path, pkg-config, then build dir."""\n') + f.write(" try:\n") + f.write(' return ctypes.CDLL("libts.so", ctypes.RTLD_GLOBAL)\n') + f.write(" except OSError:\n") + f.write(" pass\n") + f.write(" try:\n") + f.write(" import subprocess\n") + f.write(" libdir = subprocess.check_output(\n") + f.write(' ["pkg-config", "--variable=libdir", "ts"],\n') + f.write(" stderr=subprocess.DEVNULL,\n") + f.write(" ).decode().strip()\n") + f.write( + ' return ctypes.CDLL(os.path.join(libdir, "libts.so"), ctypes.RTLD_GLOBAL)\n' + ) + f.write(" except (OSError, subprocess.CalledProcessError):\n") + f.write(" pass\n") + f.write(" build_dir = os.path.join(\n") + f.write( + " os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),\n" + ) + f.write(' "build",\n') + f.write(" )\n") + f.write(' lib_path = os.path.join(build_dir, "libts.so")\n') + f.write(" if os.path.exists(lib_path):\n") + f.write(" return ctypes.CDLL(lib_path, ctypes.RTLD_GLOBAL)\n") + f.write( + ' raise OSError("Cannot find libts.so. Run make install or build from source.")\n\n\n' + ) + + f.write("__version__ = '0.1.0'\n\n") + f.write("_lib = _find_lib()\n\n") + + # --- Helpers --- + f.write("_CD = ctypes.c_double\n\n\n") + + f.write("def _cp(arr):\n") + f.write(' """Copy numpy array to ctypes buffer (tracked in TLS list)."""\n') + f.write(" buf = (_CD * len(arr))(*arr)\n") + f.write(" if not hasattr(_tls, 'bufs'):\n") + f.write(" _tls.bufs = []\n") + f.write(" _tls.bufs.append(buf)\n") + f.write(" return buf\n\n\n") + + f.write("def _clr():\n") + f.write(' """Clear tracked ctypes buffers."""\n') + f.write(" try:\n") + f.write(" if hasattr(_tls, 'bufs'):\n") + f.write(" _tls.bufs.clear()\n") + f.write(" except Exception:\n") + f.write(" pass\n\n\n") + + # --- Parameter structs --- + f.write("# --- Parameter structs ---\n") + seen_structs = set() + for fm in sorted(funcs, key=lambda x: x.name): + if fm.struct_fields and fm.name not in seen_structs: + cls_name = f"ts_{fm.name}_params_t" + f.write(f"class {cls_name}(ctypes.Structure):\n") + f.write(" _fields_ = [\n") + for c_fname, py_fname, pytype in fm.struct_fields: + f.write(f' ("{py_fname}", {pytype}),\n') + f.write(" ]\n\n") + seen_structs.add(fm.name) + + # --- FUNC_META table --- + f.write("# --- Function metadata table ---\n") + f.write( + "# Each entry: (name, x_n, y_n, struct_class_or_None, [(param_name, default), ...])\n" + ) + f.write("FUNC_META = [\n") + for fm in sorted(funcs, key=lambda x: x.name): + struct_cls = None + params = [] + if fm.struct_fields: + struct_cls = f"ts_{fm.name}_params_t" + for c_fname, py_fname, pytype in fm.struct_fields: + d = _py_default(pytype, py_fname, fm.name) + params.append(f" ('{py_fname}', {d!r}),") + f.write( + f" ({fm.name!r}, {fm.x_n}, {fm.y_n}, {struct_cls}, [{chr(10).join(params)}]),\n" + ) + f.write("]\n\n") + + # --- Single dispatch function --- + f.write("# --- Dispatch ---\n") + f.write("def _dispatch(meta, *x_arrays):\n") + f.write(' """Call a ts indicator via ctypes, given a FUNC_META entry."""\n') + f.write(" name, x_n, y_n, struct_cls, param_defaults = meta\n") + f.write(" if len(x_arrays) != x_n:\n") + f.write( + ' raise TypeError(f"Expected {x_n} input arrays, got {len(x_arrays)}")\n' + ) + f.write(" if x_n > 0:\n") + f.write(" n = x_arrays[0].shape[0]\n") + f.write(" for a in x_arrays:\n") + f.write(" if a.shape[0] != n:\n") + f.write( + ' raise ValueError(f"Array length mismatch: expected {n}, got {a.shape[0]}")\n' + ) + f.write(" else:\n") + f.write(" n = 0\n") + f.write(" y = [np.zeros(n, dtype=np.float64) for _ in range(y_n)]\n") + f.write(" try:\n") + f.write(" x_bufs = [_cp(x_arrays[i]) for i in range(x_n)]\n") + f.write( + " y_ptrs = [y[i].ctypes.data_as(ctypes.POINTER(ctypes.c_double)) for i in range(y_n)]\n" + ) + f.write(" c_func = getattr(_lib, f'ts_{name}_flat')\n") + f.write(" if struct_cls is not None:\n") + f.write(" _p = struct_cls()\n") + f.write(" for k, v in param_defaults:\n") + f.write(" setattr(_p, k, v)\n") + f.write( + " ret = c_func(*(y_ptrs + x_bufs + [ctypes.c_uint64(n), ctypes.byref(_p)]))\n" + ) + f.write(" else:\n") + f.write( + " ret = c_func(*(y_ptrs + x_bufs + [ctypes.c_uint64(n), None]))\n" + ) + f.write(" if ret != 0:\n") + f.write( + ' raise RuntimeError(f"ts_{name} failed with return code {ret}")\n' + ) + f.write(" return y[0] if y_n == 1 else tuple(y)\n") + f.write(" finally:\n") + f.write(" _clr()\n\n\n") + + # --- Helper for param override --- + f.write( + "def _dispatch_with_params(name, x_n, y_n, struct_cls, _p, x_arrays):\n" + ) + f.write(' """Call indicator with a pre-built params struct."""\n') + f.write(" if len(x_arrays) != x_n:\n") + f.write( + ' raise TypeError(f"Expected {x_n} input arrays, got {len(x_arrays)}")\n' + ) + f.write(" n = x_arrays[0].shape[0] if x_n > 0 else 0\n") + f.write(" y = [np.zeros(n, dtype=np.float64) for _ in range(y_n)]\n") + f.write(" try:\n") + f.write(" x_bufs = [_cp(x_arrays[i]) for i in range(x_n)]\n") + f.write( + " y_ptrs = [y[i].ctypes.data_as(ctypes.POINTER(ctypes.c_double)) for i in range(y_n)]\n" + ) + f.write(" c_func = getattr(_lib, f'ts_{name}_flat')\n") + f.write( + " ret = c_func(*(y_ptrs + x_bufs + [ctypes.c_uint64(n), ctypes.byref(_p)]))\n" + ) + f.write(" if ret != 0:\n") + f.write( + ' raise RuntimeError(f"ts_{name} failed with return code {ret}")\n' + ) + f.write(" return y[0] if y_n == 1 else tuple(y)\n") + f.write(" finally:\n") + f.write(" _clr()\n\n\n") + + # --- Dynamic attribute access --- + f.write("# --- Dynamic attribute access ---\n") + f.write("_FUNC_MAP = {m[0]: m for m in FUNC_META}\n\n\n") + + f.write("def __getattr__(name):\n") + f.write(' """Return a callable wrapper for indicator `name`."""\n') + f.write(" if name.startswith('_'):\n") + f.write( + ' raise AttributeError(f"module {__name__!r} has no attribute {name!r}")\n' + ) + f.write(" meta = _FUNC_MAP.get(name)\n") + f.write(" if meta is None:\n") + f.write( + ' raise AttributeError(f"{name!r} is not a known ts indicator")\n' + ) + f.write(" name_s, x_n, y_n, struct_cls, param_defaults = meta\n") + f.write(" def wrapper(*x_arrays, **kw):\n") + f.write(" for a in x_arrays:\n") + f.write(" if not isinstance(a, np.ndarray):\n") + f.write( + ' raise TypeError(f"Expected numpy.ndarray, got {type(a).__name__}")\n' + ) + f.write(" if struct_cls is not None and kw:\n") + f.write(" _p = struct_cls()\n") + f.write(" for k, v in kw.items():\n") + f.write(" if hasattr(_p, k):\n") + f.write(" setattr(_p, k, v)\n") + f.write(" else:\n") + f.write( + ' raise TypeError(f"Unknown parameter {k!r} for {name_s}")\n' + ) + f.write( + " return _dispatch_with_params(name_s, x_n, y_n, struct_cls, _p, x_arrays)\n" + ) + f.write(" return _dispatch(meta, *x_arrays)\n") + f.write(" return wrapper\n\n\n") + + # --- __all__ --- + f.write("__all__ = [m[0] for m in FUNC_META]\n") + + print(f"Generated {len(funcs)} entries -> {OUT}") + py_compile.compile(OUT, doraise=True) + + +if __name__ == "__main__": + main() diff --git a/scripts/gen_flat.py b/scripts/gen_flat.py new file mode 100644 index 0000000..c659432 --- /dev/null +++ b/scripts/gen_flat.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""Generate flat-array C wrappers for all ts functions. + +Each wrapper takes flat input/output arrays instead of pointer-of-pointers, +eliminating ctypes pointer marshaling overhead. +""" + +import argparse +import importlib.util +import os +import sys + +_spec = importlib.util.spec_from_file_location( + "ts_parser", + os.path.join(os.path.dirname(os.path.abspath(__file__)), "ts_parser.py"), +) +if _spec and _spec.loader: + _ts_parser = importlib.util.module_from_spec(_spec) + _spec.loader.exec_module(_ts_parser) + parse_all = _ts_parser.parse_all +else: + print("Error: could not load ts_parser", file=sys.stderr) + sys.exit(1) + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +OUT = os.path.join(ROOT, "lib", "ts_flat_generated.c") + + +def generate_wrappers(funcs, dry_run=False): + """Generate flat-array C wrapper functions.""" + if dry_run: + print(f"[dry-run] Would generate {len(funcs)} flat wrappers -> {OUT}") + return + + with open(OUT, "w", encoding="utf-8") as f: + f.write("/* Auto-generated flat-array wrappers. Do not edit. */\n\n") + f.write("#include \n") + f.write('#include "ts.h"\n\n') + + for fm in funcs: + safe_x_n = max(fm.x_n, 1) + safe_y_n = max(fm.y_n, 1) + + f.write(f"/* {fm.name}: x_n={fm.x_n}, y_n={fm.y_n}, c_n={fm.c_n} */\n") + f.write(f"int ts_{fm.name}_flat(\n") + + for i in range(fm.y_n): + f.write(f" fp_t *y{i},\n") + + for i in range(fm.x_n): + f.write(f" fp_t const *x{i},\n") + + f.write(f" uint_t n,\n") + f.write(f" ts_constant_t const *params\n") + f.write(f")\n{{\n") + + f.write(f" fp_t * const _y[{safe_y_n}] = {{") + f.write(", ".join(f"y{i}" for i in range(fm.y_n))) + f.write("};\n") + + f.write(f" fp_t const * const _x[{safe_x_n}] = {{") + f.write(", ".join(f"x{i}" for i in range(fm.x_n))) + f.write("};\n") + + for i in range(fm.y_n): + f.write(f" memset(y{i}, 0, n * sizeof(fp_t));\n") + + f.write( + f" return ts_{fm.name}(_y, {fm.y_n}, _x, {fm.x_n}, n, params);\n" + ) + f.write("}\n\n") + + +def main(): + parser = argparse.ArgumentParser(description="Generate flat-array C wrappers") + parser.add_argument( + "--dry-run", + action="store_true", + help="Print what would be generated without writing", + ) + args = parser.parse_args() + + funcs = parse_all(detect_x_access=True) + + if not funcs: + print("Warning: no ts_parameter declarations found.", file=sys.stderr) + sys.exit(1) + + generate_wrappers(funcs, dry_run=args.dry_run) + + if not args.dry_run: + print(f"Generated {len(funcs)} flat wrappers -> {OUT}") + + +if __name__ == "__main__": + main() diff --git a/scripts/gen_fuzz.py b/scripts/gen_fuzz.py new file mode 100644 index 0000000..a1d20e3 --- /dev/null +++ b/scripts/gen_fuzz.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +"""Generate fuzz table entries from ts indicator metadata. + +Parses lib/*.c via ts_parser to extract function names and input/output +counts, then generates a C source file with the complete fuzz table. +""" + +import argparse +import importlib.util +import os +import sys + +_spec = importlib.util.spec_from_file_location( + "ts_parser", + os.path.join(os.path.dirname(os.path.abspath(__file__)), "ts_parser.py"), +) +if _spec and _spec.loader: + _ts_parser = importlib.util.module_from_spec(_spec) + _spec.loader.exec_module(_ts_parser) + parse_all = _ts_parser.parse_all +else: + print("Error: could not load ts_parser", file=sys.stderr) + sys.exit(1) + +OUT = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "test", + "fuzz_table_generated.c", +) + + +def generate_table(funcs, dry_run=False): + """Generate the fuzz table C source file.""" + if dry_run: + print(f"[dry-run] Would generate {len(funcs)} fuzz entries -> {OUT}") + return + + with open(OUT, "w", encoding="utf-8") as f: + f.write("/* Auto-generated fuzz table. Do not edit. */\n\n") + f.write('#include "ts.h"\n\n') + f.write("static const fuzz_entry_t _fuzz_table_generated[] = {\n") + + for fm in funcs: + f.write(f" {{ts_{fm.name}, {fm.x_n}, {fm.y_n}}},\n") + + f.write("};\n\n") + f.write( + "static const uint_t _fuzz_generated_n = sizeof(_fuzz_table_generated) / sizeof(_fuzz_table_generated[0]);\n" + ) + + +def main(): + parser = argparse.ArgumentParser(description="Generate fuzz table from registry") + parser.add_argument( + "--dry-run", + action="store_true", + help="Print what would be generated without writing", + ) + args = parser.parse_args() + + funcs = parse_all(detect_x_access=True) + + if not funcs: + print("Warning: no ts_parameter declarations found.", file=sys.stderr) + sys.exit(1) + + generate_table(funcs, dry_run=args.dry_run) + + if not args.dry_run: + print(f"Generated {len(funcs)} fuzz entries -> {OUT}") + + +if __name__ == "__main__": + main() diff --git a/scripts/gen_registry.py b/scripts/gen_registry.py new file mode 100644 index 0000000..e80759e --- /dev/null +++ b/scripts/gen_registry.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +"""Generate ts_registry_generated.c with registry array and getter functions.""" + +import importlib.util +import os +import re +import sys + +_spec = importlib.util.spec_from_file_location( + "ts_parser", + os.path.join(os.path.dirname(os.path.abspath(__file__)), "ts_parser.py"), +) +if _spec and _spec.loader: + ts_parser = importlib.util.module_from_spec(_spec) + _spec.loader.exec_module(ts_parser) + parse_all = ts_parser.parse_all +else: + print("Error: could not load ts_parser", file=sys.stderr) + sys.exit(1) + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +OUT = os.path.join(ROOT, "lib", "ts_registry_generated.c") + + +def _resolve_params_size(params_size, name): + """Resolve params_size to a compile-time constant. + + If params_size is '0', keep it. If it's sizeof(ts_X_params_t), check + if the struct is visible in ts.h. If not, compute size from definition. + """ + if params_size == "0": + return "0" + + header_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "ts.h") + try: + with open(header_path, encoding="utf-8") as hf: + hcontent = hf.read() + if re.search(r"ts_" + re.escape(name) + r"_params_t", hcontent): + return params_size + except FileNotFoundError: + pass + # Struct not in header, compute size from definition + import glob + + lib_dir = os.path.join(ROOT, "lib") + for cfile in glob.glob(os.path.join(lib_dir, "*.c")): + try: + with open(cfile, encoding="utf-8") as f: + content = f.read() + except OSError: + continue + struct_m = re.search( + r"typedef\s+struct\s*\{([^}]+)\}\s+ts_" + re.escape(name) + r"_params_t", + content, + ) + if struct_m: + body = struct_m.group(1) + n_fp = len(re.findall(r"fp_t\s+", body)) + n_uint = len(re.findall(r"uint_t\s+", body)) + return str(n_fp * 8 + n_uint * 8) + return "0" + + +def main(): + funcs = parse_all(detect_x_access=False) + + if not funcs: + print("Warning: no ts_parameter declarations found.", file=sys.stderr) + sys.exit(1) + + with open(OUT, "w", encoding="utf-8") as f: + f.write("/* Auto-generated function registry. Do not edit. */\n\n") + f.write('#include "ts.h"\n\n') + f.write(f"static const uint_t _ts_fn_count = {len(funcs)};\n\n") + f.write("static const ts_fn_info_t _ts_registry[] = {\n") + for fm in funcs: + ps = _resolve_params_size(fm.params_size, fm.name) + f.write( + f" {{ .n_inputs = {fm.x_n}, .n_outputs = {fm.y_n}, " + f".n_constants = {fm.c_n}, .params_size = {ps} }},\n" + ) + f.write("};\n\n") + f.write("static const char * const _ts_fn_names[] = {\n") + for fm in funcs: + f.write(f' "{fm.name}",\n') + f.write("};\n\n") + + # Getter functions + f.write("uint_t ts_fn_count( void ) { return _ts_fn_count; }\n\n") + f.write("const ts_fn_info_t* ts_registry_get( uint_t i )\n") + f.write("{\n") + f.write(" return i < _ts_fn_count ? &_ts_registry[i] : NULL;\n") + f.write("}\n\n") + f.write("const char* ts_fn_name_get( uint_t i )\n") + f.write("{\n") + f.write(" return i < _ts_fn_count ? _ts_fn_names[i] : NULL;\n") + f.write("}\n") + + print(f"Generated {len(funcs)} entries -> {OUT}") + + +if __name__ == "__main__": + main() diff --git a/scripts/ts_parser.py b/scripts/ts_parser.py new file mode 100644 index 0000000..2ad3629 --- /dev/null +++ b/scripts/ts_parser.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""Shared parser for ts indicator metadata. + +Parses lib/*.c for ts_parameter_t declarations and extracts: +- name, n_inputs, n_outputs, n_constants +- Actual x[N] access count (may exceed declared n_inputs) +- Parameter struct fields + +All generators should import this instead of duplicating regex logic. +""" + +import glob +import os +import re +import warnings + +LIB_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "lib" +) + +SKIP_PREFIXES = ("ts_flat", "ts_registry", "gen_", "update_") +SKIP_NAMES = {"test"} + +PY_RESERVED_NAMES = {"lambda": "lambda_", "type": "type_", "id": "id_"} + +VOLUME_INDEX = 4 +OHLCV_COUNT = 5 + + +class FuncMeta: + """Metadata for a single ts indicator function.""" + + __slots__ = ("name", "x_n", "y_n", "c_n", "params_size", "struct_fields") + + def __init__(self, name, x_n, y_n, c_n, params_size="0", struct_fields=None): + self.name = name + self.x_n = x_n + self.y_n = y_n + self.c_n = c_n + self.params_size = params_size + self.struct_fields = struct_fields or [] + + +def _detect_actual_x_n(content, match_end, name): + """Detect actual x[N] access in function body.""" + remaining = content[match_end:] + func_def = re.search(rf"\bts_{name}\s*\([^)]*\)\s*{{", remaining) + if not func_def: + return 0 + brace_pos = remaining.find("{", func_def.start()) + depth = 0 + func_body = "" + for i in range(brace_pos, len(remaining)): + if remaining[i] == "{": + depth += 1 + elif remaining[i] == "}": + depth -= 1 + if depth == 0: + func_body = remaining[brace_pos : i + 1] + break + indices = [ + int(xm.group(1)) for xm in re.finditer(r"x\s*\[\s*(\d+)\s*\]", func_body) + ] + return max(indices) + 1 if indices else 0 + + +def _parse_struct_fields(content, name): + """Parse parameter struct fields for a function.""" + struct_m = re.search( + r"typedef\s+struct\s*\{([^}]+)\}\s+ts_" + re.escape(name) + r"_params_t", + content, + ) + if not struct_m: + return [] + fields_text = struct_m.group(1) + fields_text = re.sub(r"/\*.*?\*/", "", fields_text, flags=re.DOTALL) + fields_text = re.sub(r"//.*", "", fields_text) + fields = [] + for fm in re.finditer(r"^\s*(\w+)\s+(\w+)\s*;", fields_text, re.MULTILINE): + ctype, c_fname = fm.group(1), fm.group(2) + if ctype == "uint_t": + pytype = "ctypes.c_uint64" + elif ctype == "fp_t": + pytype = "ctypes.c_double" + else: + warnings.warn( + f"Unknown type '{ctype}' in struct ts_{name}_params_t, defaulting to c_double" + ) + pytype = "ctypes.c_double" + py_fname = PY_RESERVED_NAMES.get(c_fname, c_fname) + fields.append((c_fname, py_fname, pytype)) + return fields + + +def parse_all(detect_x_access=True): + """Parse all ts indicator functions from lib/*.c. + + Args: + detect_x_access: If True, detect actual x[N] access in function body + and adjust x_n accordingly. + + Returns: + List of FuncMeta objects. + """ + funcs = [] + for cfile in sorted(glob.glob(os.path.join(LIB_DIR, "*.c"))): + basename = os.path.basename(cfile) + stem = basename.replace(".c", "") + if any(stem.startswith(p) for p in SKIP_PREFIXES) or stem in SKIP_NAMES: + continue + try: + with open(cfile, encoding="utf-8") as f: + content = f.read() + except OSError as e: + warnings.warn(f"Could not read {cfile}: {e}") + continue + + for m in re.finditer( + r"ts_parameter_t\s+const\s+ts_(\w+)_p\s*=\s*\{\s*" + r"\.n_inputs\s*=\s*(\d+)\s*,\s*" + r"\.n_outputs\s*=\s*(\d+)\s*,\s*" + r"\.n_constants\s*=\s*(\d+)" + r"(?:\s*,\s*\.params_size\s*=\s*(\S+?))?\s*\}", + content, + ): + name = m.group(1) + x_n = int(m.group(2)) + y_n = int(m.group(3)) + c_n = int(m.group(4)) + params_size = m.group(5) or "0" + + if detect_x_access: + actual = _detect_actual_x_n(content, m.end(), name) + if actual > x_n: + x_n = actual + if x_n < OHLCV_COUNT and re.search( + r"\bx\s*\[\s*" + str(VOLUME_INDEX) + r"\s*\]", + content[m.end() :], + ): + x_n = OHLCV_COUNT + + struct_fields = _parse_struct_fields(content, name) if c_n > 0 else [] + + funcs.append(FuncMeta(name, x_n, y_n, c_n, params_size, struct_fields)) + + return funcs + + +def parse_names_only(): + """Parse function names and basic metadata (no x access detection). + + Returns: + Dict mapping name -> {"x_n": int, "y_n": int}. + """ + meta = {} + for fm in parse_all(detect_x_access=True): + if fm.name not in meta: + meta[fm.name] = {"x_n": fm.x_n, "y_n": fm.y_n} + return meta -- cgit v1.3.1