Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 26 additions & 3 deletions pineforge_codegen/codegen/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -2661,15 +2661,38 @@ def _collect_ta_runtime_resets(self) -> list[str]:
f"{site.member_name} = {site.class_name}({', '.join(runtime_args)});"
)

# Security-context TA copies (same ctor args as their main-context site)
# Security-context TA copies. Normally these share the ctor args of
# their main-context site, but a request.security nested in a helper
# called at several sites is cloned per call site (distinct sec_id +
# callsite_idx) while the shared TA site's ``ctor_args`` were resolved
# ONCE (against the first call site). Every clone would then size its
# indicator from call site 0's argument (e.g. four EMAs all pinned to
# crossFastLen instead of the per-site fast/slow lengths). Resolve each
# sec's ctor args against ITS call site by reusing the per-call-site
# function-clone TA remap (identity for cs0 / non-clones, so all other
# output stays byte-identical).
sec_call_by_id = {it["sec_id"]: it for it in self._security_calls}
ta_site_by_member = {s.member_name: s for s in self.ctx.ta_call_sites}
for info in self._security_eval_info:
sec_item = sec_call_by_id.get(info["sec_id"])
sec_containing = (sec_item or {}).get("containing_func") or ""
sec_cs_idx = (sec_item or {}).get("callsite_idx")
for idx, variants in (info.get("ta_variants") or {}).items():
site = self.ctx.ta_call_sites[idx]
if not site.ctor_args:
ctor_site = site
if sec_containing and sec_cs_idx is not None:
remap = self._func_cs_ta_remap.get((sec_containing, sec_cs_idx))
if remap:
cloned_name = remap.get(site.member_name)
if cloned_name and cloned_name != site.member_name:
cand = ta_site_by_member.get(cloned_name)
if cand is not None:
ctor_site = cand
if not ctor_site.ctor_args:
continue
runtime_args = []
any_runtime = False
for a in site.ctor_args:
for a in ctor_site.ctor_args:
rt = self._runtime_ctor_arg_for_reset(a)
if rt is not None:
runtime_args.append(rt)
Expand Down
124 changes: 118 additions & 6 deletions pineforge_codegen/codegen/security.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,10 +59,10 @@
from __future__ import annotations

from ..ast_nodes import (
ASTNode, Assignment, BinOp, BreakStmt, ContinueStmt, ExprStmt, ForStmt,
ForInStmt, FuncCall, FuncDef, Identifier, IfStmt, MemberAccess, NumberLiteral,
StringLiteral, Subscript, SwitchStmt, Ternary, TupleAssign, TupleLiteral,
UnaryOp, VarDecl, WhileStmt,
ASTNode, Assignment, BinOp, BoolLiteral, BreakStmt, ContinueStmt, ExprStmt,
ForStmt, ForInStmt, FuncCall, FuncDef, Identifier, IfStmt, MemberAccess,
NumberLiteral, StringLiteral, Subscript, SwitchStmt, Ternary, TupleAssign,
TupleLiteral, UnaryOp, VarDecl, WhileStmt,
)
from ..analyzer import (
FuncInfo, TACallSite, TA_MULTI_CTOR, TA_NO_CTOR, TA_PERIOD_ARG,
Expand Down Expand Up @@ -353,6 +353,116 @@ def _literal_int_for_security_index(self, node) -> int | None:
return None
return None

# In PineForge batch backtests these barstate flags are compile-time
# constants (see support_checker / codegen.visit_expr barstate emission):
# every bar is historical, none realtime/last. The other flags
# (isfirst/isnew/isconfirmed) depend on runtime tick state, so they are
# deliberately absent and leave a fold "unknown".
_SECURITY_CONST_BARSTATE = {
"isrealtime": False,
"islast": False,
"ishistory": True,
"islastconfirmedhistory": False,
}

def _fold_security_const_bool(
self,
node,
helper_binding_stack: tuple[dict[str, ASTNode], ...] | None = None,
resolving: set[str] | None = None,
) -> bool | None:
"""Fold a boolean expression to a compile-time constant, or None.

Only reduces expressions whose non-literal leaves are the batch-constant
barstate flags above; anything runtime-dependent returns None. Used to
resolve a request.security history index expressed as
``barstate.isrealtime ? 1 : 0`` and friends."""
if resolving is None:
resolving = set()
if isinstance(node, BoolLiteral):
return bool(node.value)
if (
isinstance(node, MemberAccess)
and isinstance(node.object, Identifier)
and node.object.name == "barstate"
):
return self._SECURITY_CONST_BARSTATE.get(node.member)
if isinstance(node, UnaryOp) and node.op == "not":
inner = self._fold_security_const_bool(
node.operand, helper_binding_stack, resolving
)
return None if inner is None else (not inner)
if isinstance(node, BinOp) and node.op in ("and", "or"):
left = self._fold_security_const_bool(
node.left, helper_binding_stack, resolving
)
right = self._fold_security_const_bool(
node.right, helper_binding_stack, resolving
)
if left is None or right is None:
return None
return (left and right) if node.op == "and" else (left or right)
if isinstance(node, Identifier):
bound = self._security_lookup_helper_binding(node.name, helper_binding_stack)
if bound is not None:
return self._fold_security_const_bool(
bound, helper_binding_stack, resolving
)
global_expr_map = getattr(self.ctx, "global_expr_map", {}) or {}
if node.name in global_expr_map and node.name not in resolving:
resolving.add(node.name)
out = self._fold_security_const_bool(
global_expr_map[node.name], helper_binding_stack, resolving
)
resolving.remove(node.name)
return out
return None

def _resolve_security_index_literal(
self,
node,
helper_binding_stack: tuple[dict[str, ASTNode], ...] | None = None,
resolving: set[str] | None = None,
) -> int | None:
"""Resolve a request.security TA/bar-field history index to a literal int.

Extends ``_literal_int_for_security_index`` by also resolving the index
through helper-parameter bindings and global aliases and constant-folding
a ternary whose condition is a batch-constant barstate flag (e.g.
``idxHigher = barstate.isrealtime ? 1 : 0`` -> 0). A literal index short-
circuits on the first line, so behaviour is unchanged for every already-
literal index. Returns None when the index cannot be reduced to a
compile-time literal, so the caller keeps its existing rejection."""
direct = self._literal_int_for_security_index(node)
if direct is not None:
return direct
if resolving is None:
resolving = set()
if isinstance(node, Identifier):
bound = self._security_lookup_helper_binding(node.name, helper_binding_stack)
if bound is not None:
return self._resolve_security_index_literal(
bound, helper_binding_stack, resolving
)
global_expr_map = getattr(self.ctx, "global_expr_map", {}) or {}
if node.name in global_expr_map and node.name not in resolving:
resolving.add(node.name)
out = self._resolve_security_index_literal(
global_expr_map[node.name], helper_binding_stack, resolving
)
resolving.remove(node.name)
return out
return None
if isinstance(node, Ternary):
cond = self._fold_security_const_bool(node.condition, helper_binding_stack)
if cond is None:
return None
chosen = node.true_val if cond else node.false_val
return self._resolve_security_index_literal(
chosen, helper_binding_stack, resolving
)
return None

def _collect_security_ohlc_hist_fields(self, node) -> set[str]:
"""Which security bar fields need HTF history (subscript index >= 1)."""
out: set[str] = set()
Expand Down Expand Up @@ -445,7 +555,7 @@ def walk(n):
if isinstance(n, Subscript):
site = resolve_ta_site(n.object)
if site is not None:
idx_lit = self._literal_int_for_security_index(n.index)
idx_lit = self._resolve_security_index_literal(n.index)
if idx_lit is not None and idx_lit >= 1:
site_idx = self._ta_index_by_site_id.get(id(site))
if site_idx is not None:
Expand Down Expand Up @@ -2076,7 +2186,9 @@ def _build_security_expr(
# produced the chart-tf TA instead of the confirmed HTF value.
idx = self._ta_index_by_site_id.get(id(ta_site))
sig = self._security_binding_stack_signature(helper_binding_stack)
idx_lit = self._literal_int_for_security_index(expr_node.index)
idx_lit = self._resolve_security_index_literal(
expr_node.index, helper_binding_stack
)
if idx_lit is None:
self._codegen_error(
expr_node,
Expand Down
Loading