diff --git a/pineforge_codegen/__init__.py b/pineforge_codegen/__init__.py index e676051..19727e0 100644 --- a/pineforge_codegen/__init__.py +++ b/pineforge_codegen/__init__.py @@ -4,6 +4,7 @@ from .parser import Parser from .analyzer import Analyzer from .codegen import CodeGen +from .finite_ta_length import expand_finite_choice_extrema_lengths from .pragmas import extract_pf_trace_pragmas from .support_checker import check_support_or_raise @@ -44,6 +45,7 @@ def transpile(pine_source: str, *, check_support: bool = True, filename: str = " ast = Parser(tokens, source=pine_source, filename=filename).parse() if check_support: check_support_or_raise(ast, filename=filename) + ast = expand_finite_choice_extrema_lengths(ast) ctx = Analyzer(ast, filename=filename).analyze() # Attach after analysis: pragma expressions are not part of the # program body, so the analyzer never inspects them; the codegen @@ -81,6 +83,7 @@ def transpile_full(pine_source: str, *, check_support: bool = True, ast = Parser(tokens, source=pine_source, filename=filename).parse() if check_support: check_support_or_raise(ast, filename=filename) + ast = expand_finite_choice_extrema_lengths(ast) ctx = Analyzer(ast, filename=filename).analyze() ctx.pf_trace_pragmas = pragmas gen = CodeGen(ctx) diff --git a/pineforge_codegen/finite_ta_length.py b/pineforge_codegen/finite_ta_length.py new file mode 100644 index 0000000..97b36c8 --- /dev/null +++ b/pineforge_codegen/finite_ta_length.py @@ -0,0 +1,798 @@ +"""Bounded lowering for finite-choice ``ta.highest``/``ta.lowest`` lengths. + +Pine permits a ``series int`` length for its rolling extrema. PineForge's +runtime extrema objects intentionally keep one fixed constructor length, so a +series-selected length cannot be represented by mutating or rebuilding one +object: after a shorter window has discarded history, a later longer window +cannot recover it. + +The small, exact subset handled here needs no runtime ABI change. When a +top-level extrema declaration chooses between two bar-invariant lengths, emit +one fixed extrema history for each choice, advance both on every bar, and +select the value for the current condition. For example:: + + n = regime ? fastInput : slowInput + lo = ta.lowest(low, n) + +is lowered in the AST to the semantic equivalent:: + + pf_ta_choice_1_selected = n + pf_ta_choice_1_true = ta.lowest(low, fastInput) + pf_ta_choice_1_false = ta.lowest(low, slowInput) + lo = pf_ta_choice_1_selected == fastInput ? + pf_ta_choice_1_true : pf_ta_choice_1_false + +The transform is deliberately narrow: + +* only direct top-level variable declarations; +* only ``ta.highest`` and ``ta.lowest`` positional call forms; +* exactly one ternary (inline or through a top-level length alias), with a + direct comparison operator as its condition; +* each branch must be a positive integer literal or a direct immutable + ``input.int`` alias whose declared domain is provably positive; and +* the source must be a plain identifier (the one-argument default-source form + is also accepted). + +The authored length is snapshotted exactly once at the original call site. +Aliases and every transitive selector/arm alias must be ordinary immutable +declarations: ``var``/``varip`` dependencies and any name reassigned anywhere +in the program stay unsupported. Direct inline ``input.int`` arms are also +left unsupported so lowering cannot duplicate their declarations. + +Everything else retains the existing loud unsupported-length diagnostic. In +particular, this is not a claim of arbitrary series-length support for EMA, +SMA, pivots, UDF-local sites, request.security evaluators, or unbounded +run-time lengths. +""" + +from __future__ import annotations + +import copy +from dataclasses import replace + +from .ast_nodes import ( + Assignment, + BinOp, + FuncCall, + FuncDef, + Identifier, + MemberAccess, + MethodDef, + NumberLiteral, + Program, + Subscript, + Ternary, + TupleAssign, + TupleLiteral, + UnaryOp, + VarDecl, +) + + +_EXTREMA_NAMES = frozenset({"highest", "lowest"}) +_COMPARISON_OPS = frozenset({"==", "!=", "<", "<=", ">", ">="}) +_ARITHMETIC_OPS = frozenset({"+", "-", "*", "/", "%"}) +_BAR_IDENTIFIERS = frozenset({ + "open", "high", "low", "close", "volume", + "hl2", "hlc3", "ohlc4", "hlcc4", + "time", "time_close", "bar_index", "last_bar_index", "timenow", +}) + + +def _call_name(node: FuncCall) -> tuple[str | None, str | None]: + callee = node.callee + if not isinstance(callee, MemberAccess): + return None, None + if not isinstance(callee.object, Identifier): + return None, None + return callee.object.name, callee.member + + +def _is_input_int_call(node) -> bool: + return ( + isinstance(node, FuncCall) + and _call_name(node) == ("input", "int") + ) + + +def _is_positive_int_literal(node) -> bool: + return ( + isinstance(node, NumberLiteral) + and isinstance(node.value, int) + and not isinstance(node.value, bool) + and node.value > 0 + ) + + +def _is_positive_bounded_input_int(node) -> bool: + """Whether an input owns a provably positive run-time domain.""" + + if not _is_input_int_call(node) or not node.args: + return False + if not _is_positive_int_literal(node.args[0]): + return False + + minval = node.kwargs.get("minval") + if _is_positive_int_literal(minval): + return True + + options = node.kwargs.get("options") + return ( + isinstance(options, TupleLiteral) + and bool(options.elements) + and all(_is_positive_int_literal(item) for item in options.elements) + ) + + +def _is_fixed_int_length( + node, + definitions: dict[str, tuple[int, VarDecl]], + *, + before_index: int, + ambiguous_names: frozenset[str] = frozenset(), +) -> bool: + """Whether ``node`` is one bar-invariant integer length choice. + + The only admitted leaves are a positive literal or one direct, unique, + top-level ``input.int`` alias whose entire selectable domain is positive. + This deliberately excludes alias chains and unconstrained inputs. + """ + + if _is_positive_int_literal(node): + return True + # A direct inline input call would be copied into both fixed histories. + if _is_input_int_call(node): + return False + if ( + not isinstance(node, Identifier) + or node.name in ambiguous_names + ): + return False + found = definitions.get(node.name) + if found is None: + return False + def_index, declaration = found + if def_index >= before_index or declaration.value is None: + return False + if declaration.type_hint not in (None, "int"): + return False + return _is_positive_bounded_input_int(declaration.value) + + +def _is_numeric_expr( + node, + definitions: dict[str, tuple[int, VarDecl]], + *, + before_index: int, + assigned_names: set[str], + ambiguous_names: frozenset[str], + authored_names: frozenset[str], + seen: frozenset[str] = frozenset(), +) -> bool: + """Prove the small numeric expression subset used by Wellman's regime.""" + + if isinstance(node, NumberLiteral): + return not isinstance(node.value, bool) + if isinstance(node, Identifier): + if node.name in _BAR_IDENTIFIERS: + return node.name not in authored_names + if node.name in seen or node.name in ambiguous_names: + return False + found = definitions.get(node.name) + if found is None: + return False + def_index, declaration = found + if ( + def_index >= before_index + or declaration.value is None + or declaration.is_var + or declaration.is_varip + or declaration.name in assigned_names + # Wellman's selector aliases are float-valued. An explicit int + # alias can mask a float-to-int annotation error, so integer length + # inputs are admitted only by the separate fixed-length proof. + or declaration.type_hint not in (None, "float") + ): + return False + return _is_numeric_expr( + declaration.value, + definitions, + before_index=def_index, + assigned_names=assigned_names, + ambiguous_names=ambiguous_names, + authored_names=authored_names, + seen=seen | {node.name}, + ) + if isinstance(node, UnaryOp): + return node.op in {"+", "-"} and _is_numeric_expr( + node.operand, + definitions, + before_index=before_index, + assigned_names=assigned_names, + ambiguous_names=ambiguous_names, + authored_names=authored_names, + seen=seen, + ) + if isinstance(node, BinOp): + return node.op in _ARITHMETIC_OPS and all( + _is_numeric_expr( + child, + definitions, + before_index=before_index, + assigned_names=assigned_names, + ambiguous_names=ambiguous_names, + authored_names=authored_names, + seen=seen, + ) + for child in (node.left, node.right) + ) + if isinstance(node, Ternary): + return _is_numeric_comparison( + node.condition, + definitions, + before_index=before_index, + assigned_names=assigned_names, + ambiguous_names=ambiguous_names, + authored_names=authored_names, + seen=seen, + ) and all( + _is_numeric_expr( + child, + definitions, + before_index=before_index, + assigned_names=assigned_names, + ambiguous_names=ambiguous_names, + authored_names=authored_names, + seen=seen, + ) + for child in (node.true_val, node.false_val) + ) + if isinstance(node, FuncCall): + namespace, name = _call_name(node) + if namespace == "input" and name in {"int", "float"}: + return True + # Keep the TA proof intentionally evidence-bound to Wellman's ATRs. + if not ( + namespace == "ta" + and name == "atr" + and not node.kwargs + and len(node.args) == 1 + ): + return False + length = node.args[0] + if not _is_fixed_int_length( + length, + definitions, + before_index=before_index, + ambiguous_names=ambiguous_names, + ): + return False + if isinstance(length, Identifier): + _def_index, declaration = definitions[length.name] + if ( + declaration.is_var + or declaration.is_varip + or length.name in assigned_names + ): + return False + return True + return False + + +def _is_numeric_comparison( + node, + definitions: dict[str, tuple[int, VarDecl]], + *, + before_index: int, + assigned_names: set[str], + ambiguous_names: frozenset[str], + authored_names: frozenset[str], + seen: frozenset[str] = frozenset(), +) -> bool: + """Whether ``node`` is a comparator over two proven numeric operands.""" + + return ( + isinstance(node, BinOp) + and node.op in _COMPARISON_OPS + and _is_numeric_expr( + node.left, + definitions, + before_index=before_index, + assigned_names=assigned_names, + ambiguous_names=ambiguous_names, + authored_names=authored_names, + seen=seen, + ) + and _is_numeric_expr( + node.right, + definitions, + before_index=before_index, + assigned_names=assigned_names, + ambiguous_names=ambiguous_names, + authored_names=authored_names, + seen=seen, + ) + ) + + +def _resolve_choice( + node, + definitions: dict[str, tuple[int, VarDecl]], + *, + before_index: int, + seen: frozenset[str] = frozenset(), + ambiguous_names: frozenset[str] = frozenset(), + aliases: tuple[VarDecl, ...] = (), +) -> tuple[Ternary, int, tuple[VarDecl, ...]] | None: + """Resolve one inline or top-level-aliased ternary length expression.""" + + if isinstance(node, Ternary): + return node, before_index, aliases + if ( + not isinstance(node, Identifier) + or node.name in seen + or node.name in ambiguous_names + ): + return None + found = definitions.get(node.name) + if found is None: + return None + def_index, declaration = found + if def_index >= before_index or declaration.value is None: + return None + return _resolve_choice( + declaration.value, + definitions, + before_index=def_index, + seen=seen | {node.name}, + ambiguous_names=ambiguous_names, + aliases=aliases + (declaration,), + ) + + +def _depends_on_series( + node, + definitions: dict[str, tuple[int, VarDecl]], + *, + before_index: int, + seen: frozenset[str] = frozenset(), + ambiguous_names: frozenset[str] = frozenset(), +) -> bool: + """Conservatively identify a condition that is definitely bar-varying. + + A stable input/timeframe selector was already supported by the ordinary + one-object runtime-reset route. Expanding it would be semantically valid + but would churn generated output and state for no benefit, so this pass is + enabled only when a series dependency is visible through simple top-level + aliases. + """ + + if isinstance(node, Identifier): + if node.name in _BAR_IDENTIFIERS: + return True + if node.name in seen or node.name in ambiguous_names: + return False + found = definitions.get(node.name) + if found is None: + return False + def_index, declaration = found + if def_index >= before_index or declaration.value is None: + return False + return _depends_on_series( + declaration.value, + definitions, + before_index=def_index, + seen=seen | {node.name}, + ambiguous_names=ambiguous_names, + ) + if isinstance(node, Subscript): + return True + if isinstance(node, (BinOp,)): + return _depends_on_series( + node.left, definitions, before_index=before_index, seen=seen, + ambiguous_names=ambiguous_names, + ) or _depends_on_series( + node.right, definitions, before_index=before_index, seen=seen, + ambiguous_names=ambiguous_names, + ) + if isinstance(node, UnaryOp): + return _depends_on_series( + node.operand, definitions, before_index=before_index, seen=seen, + ambiguous_names=ambiguous_names, + ) + if isinstance(node, Ternary): + return any( + _depends_on_series( + child, definitions, before_index=before_index, seen=seen, + ambiguous_names=ambiguous_names, + ) + for child in (node.condition, node.true_val, node.false_val) + ) + if isinstance(node, MemberAccess): + if isinstance(node.object, Identifier): + if node.object.name in {"barstate", "strategy"}: + return True + if node.object.name in {"input", "timeframe", "syminfo", "math"}: + return False + return _depends_on_series( + node.object, definitions, before_index=before_index, seen=seen, + ambiguous_names=ambiguous_names, + ) + if isinstance(node, FuncCall): + namespace, _name = _call_name(node) + if namespace in {"ta", "strategy", "request"}: + return True + if namespace == "input": + return False + if namespace in {"math", "timeframe"}: + return any( + _depends_on_series( + arg, definitions, before_index=before_index, seen=seen, + ambiguous_names=ambiguous_names, + ) + for arg in node.args + ) + # Bare time/session functions and other calls are not admitted merely + # because their result might be dynamic. The bounded route requires a + # dependency it can prove, keeping unknown UDF semantics unchanged. + return False + return False + + +def _plain_extrema_call(node) -> tuple[str, object, int] | None: + """Return ``(name, source, length_index)`` for the bounded call form.""" + + if not isinstance(node, FuncCall) or node.kwargs: + return None + namespace, name = _call_name(node) + if namespace != "ta" or name not in _EXTREMA_NAMES: + return None + if len(node.args) == 1: + # ta.highest(length) / ta.lowest(length): the analyzer supplies the + # native high/low source after this transform. + return name, None, 0 + if len(node.args) != 2 or not isinstance(node.args[0], Identifier): + return None + expected_source = "low" if name == "lowest" else "high" + if node.args[0].name != expected_source: + return None + return name, node.args[0], 1 + + +def _all_bound_names(program: Program) -> set[str]: + """Collect authored binders so generated top-level names cannot collide.""" + + names: set[str] = set() + + def walk(node) -> None: + if node is None: + return + if isinstance(node, VarDecl): + names.add(node.name) + if isinstance(node, (FuncDef, MethodDef)): + names.add(node.name) + params = getattr(node, "params", None) + if isinstance(params, list): + names.update(name for name in params if isinstance(name, str)) + var = getattr(node, "var", None) + if isinstance(var, str): + names.add(var) + vars_ = getattr(node, "vars", None) + if isinstance(vars_, list): + names.update(name for name in vars_ if isinstance(name, str)) + tuple_names = getattr(node, "names", None) + if isinstance(tuple_names, list): + names.update(name for name in tuple_names if isinstance(name, str)) + if not hasattr(node, "__dict__"): + return + for value in vars(node).values(): + if isinstance(value, list): + for child in value: + if isinstance(child, tuple): + for item in child: + walk(item) + else: + walk(child) + elif isinstance(value, dict): + for child in value.values(): + walk(child) + else: + walk(value) + + walk(program) + return names + + +def _assigned_names(program: Program) -> set[str]: + """Return top-level-scope names explicitly written anywhere in a bar. + + Assignments nested in control flow still mutate the surrounding Pine + variable. Callable bodies are included conservatively as well: even an + authored-invalid attempt to mutate a global dependency must not become + executable merely because this lowering ran first. + """ + + names: set[str] = set() + + def walk(node) -> None: + if node is None: + return + if isinstance(node, Assignment) and isinstance(node.target, Identifier): + names.add(node.target.name) + if isinstance(node, TupleAssign): + names.update(node.names) + if not hasattr(node, "__dict__"): + return + for value in vars(node).values(): + if isinstance(value, list): + for child in value: + if isinstance(child, tuple): + for item in child: + walk(item) + else: + walk(child) + elif isinstance(value, dict): + for child in value.values(): + walk(child) + else: + walk(value) + + for statement in program.body: + walk(statement) + return names + + +def _dependency_declarations( + node, + definitions: dict[str, tuple[int, VarDecl]], + *, + before_index: int, + ambiguous_names: frozenset[str] = frozenset(), + authored_names: frozenset[str] = frozenset(), +) -> tuple[dict[str, VarDecl], bool]: + """Resolve all visible top-level aliases transitively used by ``node``.""" + + dependencies: dict[str, VarDecl] = {} + valid = True + + def walk(current, limit: int) -> None: + nonlocal valid + if current is None: + return + if isinstance(current, Identifier): + if current.name in ambiguous_names: + valid = False + return + found = definitions.get(current.name) + if found is None: + if current.name in authored_names: + # A tuple/loop/callable/other authored binder is not a + # visible unique top-level VarDecl in this bounded model. + # It must not be mistaken for an external Pine builtin. + valid = False + return + def_index, declaration = found + if def_index >= limit: + # A known global exists, but not yet at this authored lexical + # point. Treat it as an invalid forward reference rather than + # silently compiling against a first-bar default. + valid = False + return + if current.name in dependencies: + return + dependencies[current.name] = declaration + walk(declaration.value, def_index) + return + if isinstance(current, (FuncDef, MethodDef)): + return + if not hasattr(current, "__dict__"): + return + for value in vars(current).values(): + if isinstance(value, list): + for child in value: + if isinstance(child, tuple): + for item in child: + walk(item, limit) + else: + walk(child, limit) + elif isinstance(value, dict): + for child in value.values(): + walk(child, limit) + else: + walk(value, limit) + + walk(node, before_index) + return dependencies, valid + + +def expand_finite_choice_extrema_lengths(program: Program) -> Program: + """Return ``program`` with the bounded finite-choice extrema lowering. + + The parser's original tree is left untouched. This matters because the + support checker runs on the authored program and because callers may reuse + a parsed AST in tests or diagnostics. + """ + + if not isinstance(program, Program): + return program + + definition_occurrences: dict[str, list[tuple[int, VarDecl]]] = {} + for index, statement in enumerate(program.body): + if isinstance(statement, VarDecl): + definition_occurrences.setdefault(statement.name, []).append( + (index, statement) + ) + definitions = { + name: occurrences[-1] + for name, occurrences in definition_occurrences.items() + } + ambiguous_names = frozenset( + name + for name, occurrences in definition_occurrences.items() + if len(occurrences) != 1 + ) + top_level_tuple_names = frozenset( + name + for statement in program.body + if isinstance(statement, TupleAssign) + for name in statement.names + ) + used_names = _all_bound_names(program) + authored_names = frozenset(used_names) + if authored_names.intersection({"ta", "input"} | _BAR_IDENTIFIERS): + # Namespace or bar-series spellings shadowed by any authored binder + # cannot be proven to denote Pine's native values in this pre-analysis + # pass. Fail closed for the whole bounded transform. + return program + assigned_names = _assigned_names(program) + counter = 0 + + def fresh_choice_names() -> tuple[str, str, str]: + nonlocal counter + while True: + counter += 1 + candidates = tuple( + f"pf_ta_length_choice_{counter}_{suffix}" + for suffix in ("selected", "true", "false") + ) + if not any(candidate in used_names for candidate in candidates): + used_names.update(candidates) + return candidates + + body: list = [] + for index, statement in enumerate(program.body): + if not isinstance(statement, VarDecl): + body.append(statement) + continue + if ( + statement.name in ambiguous_names + or statement.name in top_level_tuple_names + or statement.is_var + or statement.is_varip + or statement.type_hint not in (None, "float") + ): + # Rolling extrema return float; lowering must not erase an authored + # duplicate/persistent result binder or incompatible result + # annotation via the synthetic per-bar selection. + body.append(statement) + continue + + matched = _plain_extrema_call(statement.value) + if matched is None: + body.append(statement) + continue + _name, _source, length_index = matched + if _source is not None and _source.name in authored_names: + # The bounded two-argument route is only for Pine's native low or + # high series, never any authored binder shadowing that spelling. + body.append(statement) + continue + call = statement.value + length_node = call.args[length_index] + resolved_choice = _resolve_choice( + length_node, + definitions, + before_index=index, + ambiguous_names=ambiguous_names, + ) + if resolved_choice is None: + body.append(statement) + continue + choice, choice_index, choice_aliases = resolved_choice + if any( + declaration.type_hint not in (None, "int") + for declaration in choice_aliases + ): + body.append(statement) + continue + if not _is_numeric_comparison( + choice.condition, + definitions, + before_index=choice_index, + assigned_names=assigned_names, + ambiguous_names=ambiguous_names, + authored_names=authored_names, + ): + body.append(statement) + continue + if not _depends_on_series( + choice.condition, + definitions, + before_index=choice_index, + ambiguous_names=ambiguous_names, + ): + body.append(statement) + continue + if not _is_fixed_int_length( + choice.true_val, + definitions, + before_index=choice_index, + ambiguous_names=ambiguous_names, + ) or not _is_fixed_int_length( + choice.false_val, + definitions, + before_index=choice_index, + ambiguous_names=ambiguous_names, + ): + body.append(statement) + continue + + dependencies, dependencies_valid = _dependency_declarations( + length_node, + definitions, + before_index=index, + ambiguous_names=ambiguous_names, + authored_names=authored_names, + ) + if not dependencies_valid or any( + declaration.is_var + or declaration.is_varip + or name in assigned_names + for name, declaration in dependencies.items() + ): + body.append(statement) + continue + + selected_name, true_name, false_name = fresh_choice_names() + + true_args = copy.deepcopy(call.args) + false_args = copy.deepcopy(call.args) + true_args[length_index] = copy.deepcopy(choice.true_val) + false_args[length_index] = copy.deepcopy(choice.false_val) + true_call = replace(call, args=true_args, kwargs={}) + false_call = replace(call, args=false_args, kwargs={}) + true_decl = VarDecl( + name=true_name, + value=true_call, + loc=call.loc, + ) + false_decl = VarDecl( + name=false_name, + value=false_call, + loc=call.loc, + ) + selected_length_decl = VarDecl( + name=selected_name, + value=copy.deepcopy(length_node), + loc=call.loc, + ) + selected = Ternary( + condition=BinOp( + left=Identifier(name=selected_name, loc=call.loc), + op="==", + right=copy.deepcopy(choice.true_val), + loc=call.loc, + ), + true_val=Identifier(name=true_name, loc=call.loc), + false_val=Identifier(name=false_name, loc=call.loc), + loc=call.loc, + ) + + body.extend(( + selected_length_decl, + true_decl, + false_decl, + replace(statement, value=selected), + )) + + return replace(program, body=body) diff --git a/tests/test_codegen_ta_derived_length.py b/tests/test_codegen_ta_derived_length.py index 5d6a82c..5f656c2 100644 --- a/tests/test_codegen_ta_derived_length.py +++ b/tests/test_codegen_ta_derived_length.py @@ -327,24 +327,27 @@ def test_function_local_math_derived_length(): # --------------------------------------------------------------------------- -# Guardrail (negative): a ternary length whose condition depends on a -# series (ta.*) result stays rejected — it is NOT a stable scalar. +# Bounded series choice: Highest/Lowest can select between two fixed, +# input-backed histories even when the selector changes from bar to bar. +# Both histories must exist; rebuilding one fixed object would lose the older +# bars needed when the choice grows again. # --------------------------------------------------------------------------- -def test_series_dependent_ternary_length_rejected(): +def test_series_dependent_finite_choice_extrema_length_expands(): src = """//@version=6 strategy("series-dep-ternary") -normalSwingLookback = input.int(10, "Normal") -highVolSwingLookback = input.int(20, "High Vol") +normalSwingLookback = input.int(10, "Normal", minval=1) +highVolSwingLookback = input.int(20, "High Vol", minval=1) highVolThreshold = input.float(2.0, "Threshold") -volatilityRatio = ta.rma(close, 14) / ta.atr(14) +volatilityRatio = ta.atr(14) / ta.atr(20) activeSwingLookback = volatilityRatio >= highVolThreshold ? highVolSwingLookback : normalSwingLookback x = ta.lowest(low, activeSwingLookback) plot(x) """ - with pytest.raises(CompileError) as ei: - transpile(src) - assert "Unsupported TA constructor length" in str(ei.value) + cpp = transpile(src) + members = re.findall(r"(_ta_lowest_\d+)\(", cpp) + assert len(members) == 2 + assert sorted(_ctor_period(cpp, member) for member in members) == ["10", "20"] # --------------------------------------------------------------------------- @@ -417,10 +420,8 @@ def test_stable_reassigned_class_scope_length(): # --------------------------------------------------------------------------- -# Guardrail (negative): the wellmanapex shape — a reassigned scalar whose -# value depends on a SERIES (ta.rma result) — stays rejected even though -# the structure (initial VarDecl + reassigned in a ternary) superficially -# resembles the stable case above. +# Guardrail (negative): a genuinely reassigned scalar whose value depends on a +# series remains outside the bounded, expression-level finite-choice route. # --------------------------------------------------------------------------- def test_series_reassigned_ternary_length_rejected(): @@ -430,7 +431,9 @@ def test_series_reassigned_ternary_length_rejected(): highVolSwingLookback = input.int(20, "High Vol") highVolThreshold = input.float(2.0, "Threshold") volatilityRatio = ta.rma(close, 14) / ta.atr(14) -activeSwingLookback = volatilityRatio >= highVolThreshold ? highVolSwingLookback : normalSwingLookback +activeSwingLookback = normalSwingLookback +if volatilityRatio >= highVolThreshold + activeSwingLookback := highVolSwingLookback x = ta.lowest(low, activeSwingLookback) plot(x) """ @@ -610,11 +613,11 @@ def test_reset_uses_float_division_for_two_int_operands(): # and NEVER reassigned (``var int _tfSec = timeframe.in_seconds()``) # was treated as mutable persistent state. # -# The sentinel that MUST stay rejected: a length that is genuinely series- -# dynamic (wellmanapex ``activeSwingLookback`` — a ternary over a ta.atr -# ratio). Covered by ``test_series_dependent_ternary_length_rejected`` / -# ``test_series_reassigned_ternary_length_rejected`` above, plus the UDF-over- -# series and reassigned-var-body guards below. +# Arbitrary series-dynamic lengths still stay rejected. The one accepted +# exception is the bounded Highest/Lowest ternary whose leaves are fixed input +# lengths (covered above and in ``test_codegen_ta_finite_choice_length``). +# Reassigned-series, UDF-over-series, and reassigned-var-body guards below keep +# the broader constructor boundary intact. # --------------------------------------------------------------------------- @@ -764,4 +767,3 @@ def test_multi_statement_udf_length_rejected(): with pytest.raises(CompileError) as ei: transpile(src) assert "Unsupported TA constructor length" in str(ei.value) - diff --git a/tests/test_codegen_ta_finite_choice_length.py b/tests/test_codegen_ta_finite_choice_length.py new file mode 100644 index 0000000..030016f --- /dev/null +++ b/tests/test_codegen_ta_finite_choice_length.py @@ -0,0 +1,547 @@ +"""Finite-choice series lengths for ``ta.highest`` / ``ta.lowest``. + +The engine extrema ABI owns one fixed-size history per object. A Pine length +that selects between two input-qualified values is therefore lowered to two +fixed histories advanced on every bar, followed by a value selection. These +tests pin that bounded lowering and keep arbitrary series-length support out of +scope. +""" + +from __future__ import annotations + +import math +import re + +import pytest + +from pineforge_codegen import transpile, transpile_full +from pineforge_codegen.errors import CompileError +from tests._compile import compile_cpp +from tests.test_float_relational_runtime import _compile_and_run + + +def _member_periods(cpp: str, family: str) -> list[int]: + names = re.findall(rf"ta::{family}\s+(_ta_\w+_\d+)\s*;", cpp) + periods: list[int] = [] + for name in names: + match = re.search(rf"\b{re.escape(name)}\((\d+)\)", cpp) + assert match, (name, cpp) + periods.append(int(match.group(1))) + return periods + + +def test_wellman_shape_expands_both_extrema_histories() -> None: + src = """//@version=6 +strategy("wellman finite length reduction") +normalSwingLookback = input.int(10, "Normal", minval=1) +highVolSwingLookback = input.int(7, "High Vol", minval=1) +highVolThreshold = input.float(1.2, "Threshold") +regimeFastAtrLen = input.int(14, "Fast Regime ATR", minval=3) +regimeSlowAtrLen = input.int(200, "Slow Regime ATR", minval=50) +fastRegimeAtr = ta.atr(regimeFastAtrLen) +slowRegimeAtr = ta.atr(regimeSlowAtrLen) +volatilityRatio = slowRegimeAtr != 0 ? fastRegimeAtr / slowRegimeAtr : 1.0 +activeSwingLookback = volatilityRatio >= highVolThreshold ? highVolSwingLookback : normalSwingLookback +globalLowestLow = ta.lowest(low, activeSwingLookback) +globalHighestHigh = ta.highest(high, activeSwingLookback) +plot(globalLowestLow + globalHighestHigh) +""" + + cpp = transpile(src) + + assert sorted(_member_periods(cpp, "Lowest")) == [7, 10] + assert sorted(_member_periods(cpp, "Highest")) == [7, 10] + assert cpp.count("pf_ta_length_choice_1_true") > 1 + assert cpp.count("pf_ta_length_choice_1_false") > 1 + assert cpp.count("pf_ta_length_choice_1_selected") > 1 + assert cpp.count("pf_ta_length_choice_2_true") > 1 + assert cpp.count("pf_ta_length_choice_2_false") > 1 + assert cpp.count("pf_ta_length_choice_2_selected") > 1 + + # Snapshot the authored selected length once, then advance both fixed + # histories before selecting the matching result. + low_snapshot = cpp.index("pf_ta_length_choice_1_selected =") + low_true = cpp.index("pf_ta_length_choice_1_true =") + low_false = cpp.index("pf_ta_length_choice_1_false =") + low_select = cpp.index("globalLowestLow =", low_false) + assert low_snapshot < low_true < low_false < low_select + high_snapshot = cpp.index("pf_ta_length_choice_2_selected =") + high_true = cpp.index("pf_ta_length_choice_2_true =") + high_false = cpp.index("pf_ta_length_choice_2_false =") + high_select = cpp.index("globalHighestHigh =", high_false) + assert high_snapshot < high_true < high_false < high_select + + +@pytest.mark.parametrize("family,source", [("lowest", "low"), ("highest", "high")]) +@pytest.mark.parametrize("aliased", [False, True]) +def test_inline_and_aliased_choice_forms_compile( + family: str, source: str, aliased: bool +) -> None: + choice = "selected" if aliased else "(bar_index % 2 == 0 ? fastLen : slowLen)" + alias = ( + "selected = bar_index % 2 == 0 ? fastLen : slowLen\n" + if aliased else "" + ) + src = f"""//@version=6 +strategy("finite choice form") +fastLen = input.int(2, "Fast", minval=1) +slowLen = input.int(4, "Slow", minval=1) +{alias}value = ta.{family}({source}, {choice}) +plot(value) +""" + + cpp = transpile(src) + cxx_family = "Lowest" if family == "lowest" else "Highest" + assert sorted(_member_periods(cpp, cxx_family)) == [2, 4] + compile_cpp(cpp, label=f"finite_choice_{family}_{'alias' if aliased else 'inline'}") + + +def test_one_argument_default_source_form_expands() -> None: + src = """//@version=6 +strategy("finite choice default source") +a = input.int(2, "A", minval=1) +b = input.int(5, "B", minval=1) +n = close > open ? a : b +lo = ta.lowest(n) +hi = ta.highest(n) +plot(lo + hi) +""" + + cpp = transpile(src) + assert sorted(_member_periods(cpp, "Lowest")) == [2, 5] + assert sorted(_member_periods(cpp, "Highest")) == [2, 5] + assert ".compute(current_bar_.low)" in cpp + assert ".compute(current_bar_.high)" in cpp + + +def test_stable_input_selector_keeps_existing_single_object_route() -> None: + src = """//@version=6 +strategy("stable finite choice") +useShort = input.bool(true, "Use short") +shortLen = input.int(2, "Short", minval=1) +longLen = input.int(4, "Long", minval=1) +n = useShort ? shortLen : longLen +x = ta.lowest(low, n) +plot(x) +""" + + cpp = transpile(src) + assert len(_member_periods(cpp, "Lowest")) == 1 + assert "pf_ta_length_choice" not in cpp + + +def test_input_manifest_contains_only_authored_inputs() -> None: + src = """//@version=6 +strategy("finite choice manifest") +shortLen = input.int(2, "Short", minval=1) +longLen = input.int(4, "Long", minval=1) +n = close > open ? shortLen : longLen +x = ta.lowest(low, n) +plot(x) +""" + + full = transpile_full(src) + assert [item["title"] for item in full["inputs"]] == ["Short", "Long"] + assert "pf_ta_length_choice" not in repr(full["inputs"]) + + +def test_generated_names_avoid_authored_collision_and_are_deterministic() -> None: + src = """//@version=6 +strategy("finite choice collision") +pf_ta_length_choice_1_true = 123.0 +pf_ta_length_choice_1_false = 456.0 +a = input.int(2, "A", minval=1) +b = input.int(4, "B", minval=1) +n = close > open ? a : b +x = ta.lowest(low, n) +plot(x + pf_ta_length_choice_1_true + pf_ta_length_choice_1_false) +""" + + cpp = transpile(src) + assert cpp == transpile(src) + assert "pf_ta_length_choice_2_true" in cpp + assert "pf_ta_length_choice_2_false" in cpp + + +def test_generated_names_avoid_callable_and_snapshot_collisions() -> None: + src = """//@version=6 +strategy("finite choice callable collision") +pf_ta_length_choice_1_selected() => 1.0 +pf_ta_length_choice_1_true() => 2.0 +pf_ta_length_choice_1_false() => 3.0 +a = input.int(2, "A", minval=1) +b = input.int(4, "B", minval=1) +n = close > open ? a : b +x = ta.lowest(low, n) +plot(x + pf_ta_length_choice_1_selected() + pf_ta_length_choice_1_true() + pf_ta_length_choice_1_false()) +""" + + cpp = transpile(src) + assert "pf_ta_length_choice_2_selected" in cpp + assert "pf_ta_length_choice_2_true" in cpp + assert "pf_ta_length_choice_2_false" in cpp + compile_cpp(cpp, label="finite_choice_callable_collision") + + +@pytest.mark.parametrize( + "source", + [ + # The selected-length alias itself is changed before the call. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nn := bar_index % 3 + 1\nx = ta.lowest(low, n)", + # Copying the selector at the call would observe a different decision + # than the one captured when n was authored. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nflip = close > open\nn = flip ? a : b\nflip := not flip\nx = ta.lowest(low, n)", + # A later write matters on the next bar, so the mutation census spans + # the whole program rather than stopping at the call site. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.lowest(low, n)\na := 8", + # Persistent aliases have declaration-time semantics outside the + # proven ordinary per-bar alias subset. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nvar n = close > open ? a : b\nx = ta.lowest(low, n)", + # Inline input declarations would be duplicated by the fixed-history + # calls, so require authored top-level aliases for input leaves. + "x = ta.lowest(low, close > open ? input.int(2, \"A\") : input.int(4, \"B\"))", + # Choice leaves must already exist where the aliased ternary is + # declared, not merely by the later TA call. + "n = close > open ? a : b\na = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nx = ta.lowest(low, n)", + # The same source-order rule applies to the selector itself. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = flip ? a : b\nflip = close > open\nx = ta.lowest(low, n)", + # And to transitive aliases inside a selector even when another side + # of the expression visibly depends on a bar series. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nflip = close > threshold\nthreshold = input.float(1.0, \"Threshold\")\nn = flip ? a : b\nx = ta.lowest(low, n)", + # Duplicate choice/arm names are authored-invalid and must retain the + # analyzer's loud failure instead of being given last-wins semantics. + "a = input.int(2, \"A1\", minval=1)\na = input.int(3, \"A2\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nn = close < open ? b : a\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nflip = close > open\nflip = high > low\nn = flip ? a : b\nx = ta.lowest(low, n)", + # A write nested in a callable is conservatively part of the mutation + # census; otherwise lowering can mask an authored-invalid global write. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nmutate() =>\n a := 3\n a\nn = close > open ? a : b\nx = ta.lowest(low, n)", + # Only one direct input alias is admitted per branch; extra alias + # chains stay outside the Wellman-shaped subset. + "base = input.int(2, \"Base\", minval=1)\na = base\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.lowest(low, n)", + # Constructor lengths must be provably positive for every input value. + "a = input.int(2, \"A\")\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=0)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", options=[0, 2])\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.lowest(low, n)", + "n = close > open ? 0 : 4\nx = ta.lowest(low, n)", + "n = close > open ? -1 : 4\nx = ta.lowest(low, n)", + # A non-VarDecl authored binder is not an external/builtin name and + # cannot be forward-referenced through the bounded dependency model. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = flag and close > open ? a : b\n[flag, other] = [true, false]\nx = ta.lowest(low, n)", + # Explicit non-int annotations cannot be erased by synthetic lowering. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nfloat n = close > open ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nbool n = close > open ? a : b\nx = ta.lowest(low, n)", + "float a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.lowest(low, n)", + # Pine v6 has no numeric-to-bool coercion, and this bounded route only + # admits the direct comparison condition used by Wellman. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nflip = close > open\nn = flip ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = ta.crossover(close, open) ? a : b\nx = ta.lowest(low, n)", + # A comparison token alone is insufficient: both operands must belong + # to the bounded numeric expression subset. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nflag = close > open\nn = flag >= 1 ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = barstate.isconfirmed >= 1 ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nflag = input.bool(true, \"Flag\")\nn = flag >= close ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nmode = input.string(\"x\", \"Mode\")\nn = mode >= close ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nunknown() => close\nn = unknown() >= 1 ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nfast = input.int(14, \"Fast\", minval=1)\nslow = input.int(20, \"Slow\", minval=1)\nint ratio = ta.atr(fast) / ta.atr(slow)\nn = ratio >= 1.0 ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nint threshold = input.float(1.2, \"Threshold\")\nn = close >= threshold ? a : b\nx = ta.lowest(low, n)", + # Any authored binder shadowing a native bar-series spelling disables + # this bounded pass, even when the shadow is otherwise unrelated or is + # the extrema result itself. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\ntime = 1\nn = close > open ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = high > open ? a : b\nclose = ta.lowest(low, n)\nx = close", + # A direct top-level TupleAssign binder colliding with the extrema + # result is another duplicate target, regardless of source order. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\n[x, z] = [1.0, 2.0]\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.lowest(low, n)\n[x, z] = [1.0, 2.0]", + # Extrema are float-valued; incompatible result annotations must retain + # the original constructor/type failure rather than being converted by + # a synthetic ternary assignment. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nint x = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nbool x = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nstring x = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = 0.0\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.lowest(low, n)\nx = 0.0", + ], +) +def test_unsafe_or_ambiguous_finite_choices_remain_rejected( + source: str, +) -> None: + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(f'//@version=6\nstrategy("mutation guard")\n{source}\nplot(x)\n') + + +def test_aliased_length_is_snapshotted_at_call_site() -> None: + src = """//@version=6 +strategy("finite choice authored snapshot") +a = input.int(2, "A", minval=1) +b = input.int(4, "B", minval=1) +n = close > open ? a : b +x = ta.lowest(low, n) +plot(x) +""" + + cpp = transpile(src) + assert "pf_ta_length_choice_1_selected = n;" in cpp + compile_cpp(cpp, label="finite_choice_authored_snapshot") + + +@pytest.mark.parametrize("shadowed", ["close", "time"]) +def test_shadowed_numeric_builtin_does_not_take_finite_choice_route( + shadowed: str, +) -> None: + # The existing stable-length path happens to accept this authored-invalid + # comparator. This pass must decline it; fixing general annotation/type + # enforcement is deliberately outside the Wellman lowering. + src = f"""//@version=6 +strategy("finite choice builtin shadow") +a = input.int(2, "A", minval=1) +b = input.int(4, "B", minval=1) +bool {shadowed} = input.bool(true, "Shadow") +n = {shadowed} >= 1 ? a : b +x = ta.lowest(low, n) +plot(x) +""" + + cpp = transpile(src) + assert "pf_ta_length_choice" not in cpp + + +@pytest.mark.parametrize("qualifier", ["var", "varip"]) +def test_persistent_extrema_target_does_not_take_finite_choice_route( + qualifier: str, +) -> None: + src = f"""//@version=6 +strategy("finite choice persistent result") +a = input.int(2, "A", minval=1) +b = input.int(4, "B", minval=1) +n = close > open ? a : b +{qualifier} float x = ta.lowest(low, n) +plot(x) +""" + + # check_support=False reaches this lowering for varip; the ordinary public + # path rejects varip even earlier because batch mode has no intrabar ticks. + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(src, check_support=False) + + +@pytest.mark.parametrize("shadowed", ["ta", "input"]) +def test_shadowed_builtin_namespace_remains_rejected( + shadowed: str, +) -> None: + src = f"""//@version=6 +strategy("finite choice namespace shadow") +a = input.int(2, "A", minval=1) +b = input.int(4, "B", minval=1) +{shadowed} = close +n = close > open ? a : b +x = ta.lowest(low, n) +plot(x) +""" + + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(src) + + +def test_positive_options_domains_are_admitted() -> None: + src = """//@version=6 +strategy("finite choice positive options") +a = input.int(2, "A", options=[1, 2]) +b = input.int(4, "B", options=[3, 4]) +n = close > open ? a : b +x = ta.lowest(low, n) +plot(x) +""" + + cpp = transpile(src) + assert sorted(_member_periods(cpp, "Lowest")) == [2, 4] + compile_cpp(cpp, label="finite_choice_positive_options") + + +def test_explicit_int_choice_and_arm_aliases_are_admitted() -> None: + src = """//@version=6 +strategy("finite choice explicit int") +int a = input.int(2, "A", minval=1) +int b = input.int(4, "B", minval=1) +int n = close > open ? a : b +x = ta.lowest(low, n) +plot(x) +""" + + cpp = transpile(src) + assert sorted(_member_periods(cpp, "Lowest")) == [2, 4] + compile_cpp(cpp, label="finite_choice_explicit_int") + + +def test_explicit_float_extrema_result_is_admitted() -> None: + src = """//@version=6 +strategy("finite choice explicit float result") +int a = input.int(2, "A", minval=1) +int b = input.int(4, "B", minval=1) +int n = close > open ? a : b +float x = ta.lowest(low, n) +plot(x) +""" + + cpp = transpile(src) + assert sorted(_member_periods(cpp, "Lowest")) == [2, 4] + compile_cpp(cpp, label="finite_choice_explicit_float_result") + + +@pytest.mark.parametrize( + "source", + [ + # Arbitrary, unbounded series length. + "n = ta.barssince(close > open)\nx = ta.lowest(low, n)", + # One choice is itself series-valued rather than a fixed leaf. + "a = input.int(2, \"A\")\nn = close > open ? a : int(ta.atr(3))\nx = ta.lowest(low, n)", + # Other TA families retain their simple/fixed-length contract. + "a = input.int(2, \"A\")\nb = input.int(4, \"B\")\nn = close > open ? a : b\nx = ta.ema(close, n)", + # Non-identifier source expressions stay outside the bounded route. + "a = input.int(2, \"A\")\nb = input.int(4, \"B\")\nn = close > open ? a : b\nx = ta.lowest(low + 0.0, n)", + # The two-argument route is deliberately native-source exact: + # lowest(low, n) and highest(high, n), with no aliases or shadows. + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.lowest(close, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nn = close > open ? a : b\nx = ta.highest(open, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nsrc = low\nn = close > open ? a : b\nx = ta.lowest(src, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\nlow = close\nn = close > open ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\n[low, other] = [close, open]\nn = close > open ? a : b\nx = ta.lowest(low, n)", + "a = input.int(2, \"A\", minval=1)\nb = input.int(4, \"B\", minval=1)\n[high, other] = [close, open]\nn = close > open ? a : b\nx = ta.highest(high, n)", + ], +) +def test_unbounded_or_out_of_scope_lengths_remain_rejected(source: str) -> None: + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(f'//@version=6\nstrategy("negative")\n{source}\nplot(x)\n') + + +_RUNTIME_PINE = """//@version=6 +strategy("finite choice runtime") +fastLen = input.int(2, "Fast", minval=1) +slowLen = input.int(4, "Slow", minval=1) +selected = bar_index % 2 == 0 ? fastLen : slowLen +lo = ta.lowest(low, selected) +hi = ta.highest(high, selected) + +var float lo2 = na +var float hi2 = na +var float lo3 = na +var float hi3 = na +var float lo4 = na +var float hi4 = na +var float lo5 = na +var float hi5 = na +if bar_index == 2 + lo2 := lo + hi2 := hi +if bar_index == 3 + lo3 := lo + hi3 := hi +if bar_index == 4 + lo4 := lo + hi4 := hi +if bar_index == 5 + lo5 := lo + hi5 := hi +""" + + +_RUNTIME_DRIVER = r""" +#include +#include + +int main() { + const double highs[] = {10, 12, 11, 15, 13, 14}; + const double lows[] = {5, 4, 6, 3, 7, 2}; + Bar bars[6]; + for (int i = 0; i < 6; ++i) { + const double mid = (highs[i] + lows[i]) / 2.0; + bars[i] = Bar{mid, highs[i], lows[i], mid, 1.0, + static_cast(i) * 900000}; + } + + GeneratedStrategy strategy; + strategy.run(bars, 6); + std::cout << std::setprecision(17) + << strategy.lo2 << '\t' << strategy.hi2 << '\t' + << strategy.lo3 << '\t' << strategy.hi3 << '\t' + << strategy.lo4 << '\t' << strategy.hi4 << '\t' + << strategy.lo5 << '\t' << strategy.hi5 << '\n'; + return 0; +} +""" + + +def test_runtime_switches_between_fully_advanced_fixed_histories() -> None: + cpp = transpile(_RUNTIME_PINE) + stdout = _compile_and_run(cpp + _RUNTIME_DRIVER) + observed = tuple(float(value) for value in stdout.strip().split("\t")) + assert observed == (4.0, 12.0, 3.0, 15.0, 3.0, 15.0, 2.0, 15.0) + assert all(math.isfinite(value) for value in observed) + + +_LONG_INACTIVE_PINE = """//@version=6 +strategy("finite choice long inactive runtime") +shortLen = input.int(2, "Short", minval=1) +longLen = input.int(5, "Long", minval=1) +selected = bar_index < 6 ? shortLen : longLen +lo = ta.lowest(low, selected) +reverseSelected = bar_index < 6 ? longLen : shortLen +reverseLo = ta.lowest(low, reverseSelected) +equalSelected = bar_index % 2 == 0 ? 3 : 3 +equalLo = ta.lowest(low, equalSelected) + +var float early0 = na +var float early1 = na +var float beforeFlip = na +var float afterFlip = na +var float reverseAfterFlip = na +var float equalAfterFlip = na +if bar_index == 0 + early0 := lo +if bar_index == 1 + early1 := lo +if bar_index == 5 + beforeFlip := lo +if bar_index == 6 + afterFlip := lo + reverseAfterFlip := reverseLo + equalAfterFlip := equalLo +""" + + +_LONG_INACTIVE_DRIVER = r""" +#include +#include + +int main() { + const double lows[] = {9, 8, 1, 7, 6, 5, 4}; + Bar bars[7]; + for (int i = 0; i < 7; ++i) { + bars[i] = Bar{lows[i] + 1.0, lows[i] + 2.0, lows[i], + lows[i] + 1.0, 1.0, + static_cast(i) * 900000}; + } + + GeneratedStrategy strategy; + strategy.run(bars, 7); + std::cout << std::setprecision(17) + << strategy.early0 << '\t' << strategy.early1 << '\t' + << strategy.beforeFlip << '\t' << strategy.afterFlip << '\t' + << strategy.reverseAfterFlip << '\t' + << strategy.equalAfterFlip << '\n'; + return 0; +} +""" + + +def test_runtime_keeps_long_inactive_bank_warm_before_late_flip() -> None: + cpp = transpile(_LONG_INACTIVE_PINE) + stdout = _compile_and_run(cpp + _LONG_INACTIVE_DRIVER) + observed = tuple(float(value) for value in stdout.strip().split("\t")) + # Pin the engine's first-bar na warmup and first available value. After six + # short-window bars, switching to the long bank immediately sees its + # preserved five-bar history. + assert math.isnan(observed[0]) + assert observed[1:] == (8.0, 5.0, 1.0, 4.0, 4.0)