1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
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
|