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/ts_parser.py | 159 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 scripts/ts_parser.py (limited to 'scripts/ts_parser.py') 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