Spaces:

bryan273
/

arfox-ai

Sleeping

arfox-ai / utils.py
feat: Add application code and models via Git LFS
c072ec7 over 1 year ago
65.7 kB
import math
from typing import TypeVar, List, Tuple, Any, Union, Dict # Using Union for as_bool input
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict
from tqdm.notebook import tqdm
import requests
import time

# Generic Type Variable
T = TypeVar('T')

def last(a: List[T]) -> T:
    """Returns the last element of a list."""
    return a[-1]

# Pine rule: in Boolean expressions  na  is treated as false
def as_bool(v: Union[float, int, bool, None]) -> bool:
    """Converts a value to boolean, treating None or NaN as False."""
    if v is None or (isinstance(v, float) and math.isnan(v)):
        return False
    return bool(v)

# Helper functions for min/max emulating JavaScript's Math.min/max with NaN behavior
# JS Math.min(NaN, 5) -> 5 (if only one NaN) or NaN (if all NaN or multiple args with one NaN)
# JS Math.min(...[NaN, 5]) -> NaN
# The TS code uses `Math.max(...array)`, which means if any element in `array` is NaN, the result is NaN.

def _js_style_list_min(values: List[float]) -> float:
    """Emulates Math.min(...array) which returns NaN if any element in array is NaN."""
    if not values:
        return math.nan # Or based on specific requirement for empty list
    has_nan = False
    for val in values:
        if math.isnan(val):
            has_nan = True
            break
    if has_nan:
        return math.nan
    return min(values) if values else math.nan

def _js_style_list_max(values: List[float]) -> float:
    """Emulates Math.max(...array) which returns NaN if any element in array is NaN."""
    if not values:
        return math.nan
    has_nan = False
    for val in values:
        if math.isnan(val):
            has_nan = True
            break
    if has_nan:
        return math.nan
    return max(values) if values else math.nan

def _js_math_max(a: float, b: float) -> float:
    """Emulates JS Math.max(a,b) behavior with NaNs (prefers non-NaN)."""
    if math.isnan(a): return b
    if math.isnan(b): return a
    return max(a, b)

def _js_math_min(a: float, b: float) -> float:
    """Emulates JS Math.min(a,b) behavior with NaNs (prefers non-NaN)."""
    if math.isnan(a): return b
    if math.isnan(b): return a
    return min(a, b)

# /* ───────── basic rolling helpers ───────── */
def rolling_mean(src: List[float], length: int) -> List[float]:
    """Calculates the rolling mean (Simple Moving Average)."""
    if not src or length <= 0:
        return [math.nan] * len(src)
    out = [math.nan] * len(src)
    acc = 0.0
    for i in range(len(src)):
        if not math.isnan(src[i]): # Accumulate if not NaN
            acc += src[i]
        else: # If src[i] is NaN, the sum effectively becomes NaN for this window until enough non-NaNs flush it out or it's handled.
              # To match TS, if src[i] is NaN, acc will also become NaN if not handled.
              # The TS code doesn't check for NaN in src[i] during accumulation. acc += NaN -> acc is NaN.
              # Python: acc += float('nan') -> acc is nan. This matches.
              acc += src[i] # Allow NaN to propagate into acc

        if i >= length:
            # acc -= src[i - length];
            # If src[i-length] was NaN, acc could already be NaN. Or acc is num, src[i-length] is NaN. num - NaN = NaN.
            acc -= src[i - length] # Allow NaN propagation

        if i < length - 1:
            out[i] = math.nan
        else:
            if math.isnan(acc): # if accumulator is NaN (due to NaN in src)
                out[i] = math.nan
            else:
                out[i] = acc / length
    return out

def rolling_max(src: List[float], length: int) -> List[float]:
    """Calculates the rolling maximum."""
    if not src or length <= 0:
        return [math.nan] * len(src)
    out = [math.nan] * len(src)
    for i in range(len(src)):
        start_index = max(0, i - length + 1)
        window = src[start_index : i + 1]
        out[i] = _js_style_list_max(window)
    return out

def rolling_min(src: List[float], length: int) -> List[float]:
    """Calculates the rolling minimum."""
    if not src or length <= 0:
        return [math.nan] * len(src)
    out = [math.nan] * len(src)
    for i in range(len(src)):
        start_index = max(0, i - length + 1)
        window = src[start_index : i + 1]
        out[i] = _js_style_list_min(window)
    return out

def rolling_std(src: List[float], length: int) -> List[float]:
    """Calculates the rolling standard deviation with ddof=1."""
    if not src or length <= 1: # std requires at least 2 points for ddof=1
        return [math.nan] * len(src)

    out = [math.nan] * len(src)
    for i in range(len(src)):
        if i < length - 1:
            out[i] = math.nan
            continue

        window = src[i - length + 1 : i + 1]

        # Check for NaNs in window, if any, mean and std dev are NaN
        if any(math.isnan(x) for x in window):
            out[i] = math.nan
            continue

        m = sum(window) / length
        variance_sum = sum((x - m) ** 2 for x in window)

        # ddof = 1 means (length - 1) in denominator
        if length - 1 == 0: # Should be caught by length <= 1 check earlier
            out[i] = math.nan
        else:
            variance = variance_sum / (length - 1)
            out[i] = math.sqrt(variance)
    return out

# /* ───────── Wilder RMA & EMA ───────── */
def rma(src: List[float], length: int) -> List[float]:
    """Calculates Wilder's Recursive Moving Average."""
    if not src: return []
    if length <= 0: return [math.nan] * len(src)

    alpha = 1.0 / length
    out = [math.nan] * len(src)

    i0 = -1
    for idx, val in enumerate(src):
        if not math.isnan(val):
            i0 = idx
            break

    if i0 == -1: # All NaNs in src
        return [math.nan] * len(src)

    out[i0] = src[i0]

    for i in range(i0): # Forward-fill for any NaN before the seed
        out[i] = out[i0]

    for i in range(i0 + 1, len(src)):
        v = src[i]
        if math.isnan(v):
            out[i] = out[i-1]
        else:
            # If out[i-1] is NaN (e.g. from a long series of NaNs in src not covered by forward fill), result is NaN
            out[i] = alpha * v + (1.0 - alpha) * out[i-1]
    return out

def ema(src: List[float], length: int) -> List[float]:
    """Calculates the Exponential Moving Average."""
    if not src: return []
    if length <= 0: return [math.nan] * len(src) # Or other handling for invalid length

    k = 2.0 / (length + 1)
    out = [math.nan] * len(src)

    if not src: return [] # Should be caught already

    out[0] = src[0] # First EMA is the first source value (propagates NaN if src[0] is NaN)

    for i in range(1, len(src)):
        # If src[i] is NaN, or out[i-1] is NaN, the result will be NaN.
        out[i] = k * src[i] + (1.0 - k) * out[i-1]
    return out

# /* ───────── Wilder ATR ───────── */
def wilder_atr(high: List[float], low: List[float], close: List[float], length: int = 14) -> List[float]:
    """Calculates Wilder's Average True Range."""
    if not close or not high or not low: return []
    if not (len(close) == len(high) == len(low)):
        raise ValueError("Input lists must have the same length for ATR.")

    tr = [math.nan] * len(close)
    for i in range(len(close)):
        prev_close = close[i-1] if i > 0 else close[i]

        h_val, l_val, c_val = high[i], low[i], close[i] # Current values
        pc_val = prev_close # Previous close

        # If any component is NaN, the terms become NaN. max(NaN, num, num) is NaN.
        term1 = h_val - l_val
        term2 = abs(h_val - pc_val) if not math.isnan(h_val) and not math.isnan(pc_val) else math.nan
        term3 = abs(l_val - pc_val) if not math.isnan(l_val) and not math.isnan(pc_val) else math.nan

        if math.isnan(term1) or math.isnan(term2) or math.isnan(term3):
            tr[i] = math.nan
        else:
            tr[i] = max(term1, term2, term3)

    return rma(tr, length)

# /* ───────── Wilder RSI ───────── */
def wilder_rsi(close: List[float], length: int = 14) -> List[float]:
    """Calculates Wilder's Relative Strength Index."""
    if not close: return []
    if length <= 0: return [math.nan] * len(close)

    diff = [0.0] * len(close)
    for i in range(len(close)):
        if i > 0:
            # If close[i] or close[i-1] is NaN, diff[i] becomes NaN.
            diff[i] = close[i] - close[i-1]
        # else diff[i] is 0.0 (already initialized)

    # up/dn will propagate NaN if diff[i] is NaN. Math.max(NaN, 0) is NaN in JS, but max(NaN,0) in Python is 0 or error.
    # TS: Math.max(v, 0) -> if v is NaN, result is NaN.
    up = [(_js_math_max(d, 0.0)) if not math.isnan(d) else math.nan for d in diff]
    dn = [(_js_math_max(-d, 0.0)) if not math.isnan(d) else math.nan for d in diff]

    # The TS logic for seedU/seedD and restU/restD is specific.
    rm_up = rolling_mean(up, length)
    rm_dn = rolling_mean(dn, length)

    # .slice(0, len) in TS
    seed_u = rm_up[:length]
    seed_d = rm_dn[:length]

    rest_u_input = up[length:]
    rest_d_input = dn[length:]

    rest_u = rma(rest_u_input, length)
    rest_d = rma(rest_d_input, length)

    u_rma_list = seed_u + rest_u
    d_rma_list = seed_d + rest_d

    # Ensure lengths match original close length due to concat
    # If len(close) < length, seed_u/d might be shorter than length. rest_u/d will be from empty or short list.
    # The resulting u_rma_list / d_rma_list should naturally align with len(close).
    # Example: close len 5, length 10. up len 5. rm_up len 5 (all nan). seed_u = rm_up[:5] = 5 nans.
    # rest_u_input = up[10:] = []. rma([], 10) = []. u_rma_list = 5 nans. Correct.

    rsi_values = [math.nan] * len(close)

    for i in range(len(u_rma_list)):
        # Guard against d_rma_list being unexpectedly shorter if logic error, though it shouldn't be.
        if i >= len(d_rma_list):
            rsi_values[i] = math.nan
            continue

        val_u = u_rma_list[i]
        val_d = d_rma_list[i]

        if math.isnan(val_u) or math.isnan(val_d):
            rsi_values[i] = math.nan
        elif val_d == 0:
            if val_u == 0: # Both avg_gain and avg_loss are 0
                rsi_values[i] = math.nan # As per formula v/dRma[i] -> NaN/0 -> NaN. Some RSI define this as 50 or 100. Sticking to formula.
            else: # val_u > 0 (non-negative due to max(v,0)) and val_d == 0
                rsi_values[i] = 100.0
        else: # val_d is not 0, and neither val_u nor val_d is NaN
            rs = val_u / val_d
            rsi_values[i] = 100.0 - (100.0 / (1.0 + rs))

    return rsi_values

# /* ───────── WVF (FoxPro) – returns [last, upper, rangeHi] ───────── */
def foxpro_wvf(
    close: List[float], low: List[float],
    pd_: int = 22, bbl: int = 20, mult: float = 2.0,
    lb: int = 50, ph: float = 0.85
) -> Tuple[float, float, float]:
    """Calculates Williams VIX Fix components."""
    if not close or not low or not (len(close) == len(low)):
        return (math.nan, math.nan, math.nan)
    if len(close) == 0: return (math.nan, math.nan, math.nan)


    hi_pd = rolling_max(close, pd_)

    wvf = [math.nan] * len(close)
    for i in range(len(close)):
        # Ensure hi_pd[i] is not NaN and not zero before division
        if not math.isnan(hi_pd[i]) and hi_pd[i] != 0 and \
           not math.isnan(low[i]): # close[i] is not used in this specific formula line from TS
            wvf[i] = ((hi_pd[i] - low[i]) / hi_pd[i]) * 100.0
        else:
            wvf[i] = math.nan

    s_dev_raw = rolling_std(wvf, bbl)
    s_dev = [s * mult if not math.isnan(s) else math.nan for s in s_dev_raw]

    mid = rolling_mean(wvf, bbl)

    upper = [(m + s_dev[i]) if not math.isnan(m) and i < len(s_dev) and not math.isnan(s_dev[i]) else math.nan
             for i, m in enumerate(mid)]

    rng_hi_raw = rolling_max(wvf, lb)
    rng_hi = [v * ph if not math.isnan(v) else math.nan for v in rng_hi_raw]

    n_idx = len(wvf) - 1
    if n_idx < 0: # Empty wvf, should not happen if close is not empty
        return (math.nan, math.nan, math.nan)

    # Return last values of the calculated series
    # Ensure lists are not empty before accessing last element
    last_wvf = wvf[n_idx] if wvf else math.nan
    last_upper = upper[n_idx] if upper else math.nan
    last_rng_hi = rng_hi[n_idx] if rng_hi else math.nan

    return (last_wvf, last_upper, last_rng_hi)

# /* ───────── MA labels ───────── */
def ma_labels(
    row8: float, row13: float, row21: float,
    prev8: float, prev13: float, prev21: float
) -> str:
    """Determines MA-based market label."""
    # NaN comparisons (e.g. math.nan > 10) are False. This naturally handles NaNs in conditions.
    if row8 > row13 and row13 > row21: return 'Bullish'
    if row8 < row13 and row13 < row21: return 'Bearish'
    if prev8 > prev13 and prev13 > prev21 and row13 > row8: return 'Spec. Bearish'
    if prev8 < prev13 and prev13 < prev21 and row13 < row8: return 'Spec. Bullish'
    return 'Neutral'

# /* ───────── RSI label (same wording) ───────── */
def rsi_label(rsi: float, trend_bull: bool) -> str:
    """Determines RSI-based market label."""
    if math.isnan(rsi):
        return f"Neutral (NaN)" # Or specific NaN label

    rsi_str = f"{rsi:.1f}"

    if rsi > 85: return f"Spec Sell ({rsi_str})"
    if rsi > 80 and not trend_bull: return f"Spec Sell ({rsi_str})"
    if rsi > 70: return f"Overbought ({rsi_str})"
    if rsi < 20 and trend_bull: return f"Spec Buy ({rsi_str})"
    if rsi < 26: return f"Oversold ({rsi_str})"
    if trend_bull and rsi > 50: return f"Bullish ({rsi_str})"
    if not trend_bull and rsi < 50: return f"Bearish ({rsi_str})"
    return f"Neutral ({rsi_str})"

# /* ───────── ATR trailing stop ───────── */
def atr_trail(
    close: List[float], high: List[float], low: List[float],
    atr_p: int = 5, hhv_p: int = 10, mult: float = 2.5
) -> List[float]:
    """Calculates ATR Trailing Stop."""
    if not close or not high or not low: return []
    if not (len(close) == len(high) == len(low)):
        raise ValueError("Input lists must have the same length for ATR Trail.")

    atr_values = wilder_atr(high, low, close, atr_p)

    prev_raw = [(h_val - mult * atr_val) if not math.isnan(h_val) and not math.isnan(atr_val) else math.nan
                for h_val, atr_val in zip(high, atr_values)]

    prev = rolling_max(prev_raw, hhv_p) # Max of (high - mult * atr) over hhvP

    ts = [math.nan] * len(close)

    for i in range(len(close)):
        current_close = close[i]
        prev_val_i = prev[i]

        if i < 16:
            ts[i] = current_close
        else: # i >= 16
            # Handle NaNs for comparison: nan > x is false. x > nan is false.
            # So if prev_val_i is NaN, current_close > prev_val_i is false.
            # If current_close is NaN, current_close > prev_val_i is false.
            if not math.isnan(current_close) and not math.isnan(prev_val_i) and current_close > prev_val_i:
                ts[i] = prev_val_i
            else: # Covers current_close <= prev_val_i OR any involved value is NaN
                  # The original TS: `i ? ts[i-1] : close[i]`. Since i >= 16, `i` is true. So `ts[i-1]`.
                if i > 0:
                    ts[i] = ts[i-1]
                else: # This case (i=0 and i>=16) is impossible. Defensive.
                    ts[i] = current_close
    return ts

# /* ───────── simple SuperTrend (returns [line, trendArr]) ───────── */
def super_trend(
    close: List[float], high: List[float], low: List[float],
    length: int = 10, mult: float = 3.0
) -> Tuple[List[float], List[int]]:
    """Calculates SuperTrend indicator."""
    n = len(close)
    if n == 0 or not (n == len(high) == len(low)):
        return ([], [])

    atr_values = wilder_atr(high, low, close, length)

    hl2 = [(h_val + l_val) / 2.0 if not math.isnan(h_val) and not math.isnan(l_val) else math.nan
           for h_val, l_val in zip(high, low)]

    basic_up = [(val_hl2 - mult * val_atr) if not math.isnan(val_hl2) and not math.isnan(val_atr) else math.nan
                for val_hl2, val_atr in zip(hl2, atr_values)]
    basic_dn = [(val_hl2 + mult * val_atr) if not math.isnan(val_hl2) and not math.isnan(val_atr) else math.nan
                for val_hl2, val_atr in zip(hl2, atr_values)]

    f_up = [math.nan] * n
    f_dn = [math.nan] * n
    trend = [0] * n # 1 for uptrend, -1 for downtrend

    if n == 0: return ([], []) # Should be caught

    f_up[0] = basic_up[0]
    f_dn[0] = basic_dn[0]
    trend[0] = 1 # Seed with uptrend

    for i in range(1, n):
        prev_close_val = close[i-1]
        prev_f_up_val = f_up[i-1]
        prev_f_dn_val = f_dn[i-1]

        # Final Upper Band
        # TS: close[i-1] <= fUp[i-1] ? basicUp[i] : Math.max(basicUp[i], fUp[i-1])
        # If prev_close_val or prev_f_up_val is NaN, condition `prev_close_val <= prev_f_up_val` is False.
        if not math.isnan(prev_close_val) and not math.isnan(prev_f_up_val) and prev_close_val <= prev_f_up_val:
            f_up[i] = basic_up[i]
        else:
            f_up[i] = _js_math_max(basic_up[i], prev_f_up_val) # Emulates JS Math.max

        # Final Lower Band
        # TS: close[i-1] >= fDn[i-1] ? basicDn[i] : Math.min(basicDn[i], fDn[i-1])
        if not math.isnan(prev_close_val) and not math.isnan(prev_f_dn_val) and prev_close_val >= prev_f_dn_val:
            f_dn[i] = basic_dn[i]
        else:
            f_dn[i] = _js_math_min(basic_dn[i], prev_f_dn_val) # Emulates JS Math.min

        # Trend determination
        current_close_val = close[i]
        trend_changed = False
        if trend[i-1] == -1:
            # close[i] > fDn[i-1] (use prev_f_dn_val for fDn[i-1])
            if not math.isnan(current_close_val) and not math.isnan(prev_f_dn_val) and current_close_val > prev_f_dn_val:
                trend[i] = 1
                trend_changed = True
        elif trend[i-1] == 1:
            # close[i] < fUp[i-1] (use prev_f_up_val for fUp[i-1])
            if not math.isnan(current_close_val) and not math.isnan(prev_f_up_val) and current_close_val < prev_f_up_val:
                trend[i] = -1
                trend_changed = True

        if not trend_changed:
            trend[i] = trend[i-1]

    st_line = [math.nan] * n
    for i in range(n):
        if trend[i] == 1:
            st_line[i] = f_up[i]
        elif trend[i] == -1:
            st_line[i] = f_dn[i]
        # else trend[i] == 0 (only for first element if n=1 and not updated), st_line[i] remains math.nan

    return (st_line, trend)

# /* ───────── MACD (returns [line, signal, hist]) ───────── */
def macd_calc(src: List[float]) -> Tuple[List[float], List[float], List[float]]: # Renamed from macd to macd_calc
    """Calculates MACD, Signal Line, and Histogram."""
    if not src: return ([], [], [])

    fast_ema = ema(src, 12)
    slow_ema = ema(src, 26)

    macd_line = [(f - s) if not math.isnan(f) and not math.isnan(s) else math.nan
                 for f, s in zip(fast_ema, slow_ema)]

    signal_line = ema(macd_line, 9)

    histogram = [(m - s) if not math.isnan(m) and not math.isnan(s) else math.nan
                 for m, s in zip(macd_line, signal_line)]

    return (macd_line, signal_line, histogram)

# /* ───────── Stochastic %K  (fast) ───────── */
def _stoch_k(
    close: List[float], high: List[float], low: List[float],
    length: int = 14
) -> List[float]:
    """Helper to calculate Stochastic %K."""
    n = len(close)
    if n == 0 or length <= 0 or not (n == len(high) == len(low)):
        return [math.nan] * n

    k_values = [math.nan] * n
    for i in range(n):
        start_index = max(0, i - length + 1)

        # Use _js_style_list_min/max for consistency with TS Math.min/max(...slice)
        window_low = low[start_index : i + 1]
        window_high = high[start_index : i + 1]

        lo = _js_style_list_min(window_low)
        hi = _js_style_list_max(window_high)

        current_close = close[i]

        if math.isnan(lo) or math.isnan(hi) or math.isnan(current_close):
            k_values[i] = math.nan
        elif hi == lo: # Both are same non-NaN value, implies hi-lo is 0
            k_values[i] = 50.0 # As per TS logic
        else:
            # hi - lo cannot be zero here
            k_values[i] = (100.0 * (current_close - lo)) / (hi - lo)

    return k_values

# /* ───────── Stoch K/D  (uses the helper above) ───────── */
def stoch_kd(
    close: List[float], high: List[float], low: List[float],
    length: int = 14 # This is %K period
) -> Tuple[List[float], List[float]]:
    """Calculates Stochastic %K and %D."""
    # %D period is typically 3 for rollingMean of K
    k = _stoch_k(close, high, low, length)
    d = rolling_mean(k, 3) # %D is SMA of %K
    return (k, d)

# /* ───────── DMI (only +DI, −DI, ADX) ───────── */
def dmi_calc( # Renamed from dmi to dmi_calc
    high: List[float], low: List[float], close: List[float],
    length: int = 14
) -> Tuple[List[float], List[float], List[float]]:
    """Calculates Directional Movement Index (+DI, -DI, ADX)."""
    n = len(high)
    if n == 0 or length <= 0 or not (n == len(low) == len(close)):
        nan_list = [math.nan] * n
        return (nan_list, nan_list, nan_list) if n > 0 else ([],[],[])

    up_move = [math.nan] * n
    dn_move = [math.nan] * n

    for i in range(n):
        if i > 0:
            # NaN propagation: if high[i] or high[i-1] is NaN, up_move[i] is NaN.
            up_move[i] = high[i] - high[i-1]
            dn_move[i] = low[i-1] - low[i]
        else: # TS: up/dn are 0 for i=0.
            up_move[i] = 0.0
            dn_move[i] = 0.0

    plus_dm = [0.0] * n # Initialized to 0.0 as per TS fallback
    minus_dm = [0.0] * n

    for i in range(n):
        u = up_move[i]
        d = dn_move[i]
        # Comparisons with NaN (e.g. NaN > 0) are False.
        # So if u or d is NaN, conditions fail, and plus_dm/minus_dm remain 0 for that index.
        if not math.isnan(u) and not math.isnan(d) and u > d and u > 0:
            plus_dm[i] = u
        # else: plus_dm[i] remains 0.0 (already initialized)

        if not math.isnan(d) and not math.isnan(u) and d > u and d > 0:
            minus_dm[i] = d
        # else: minus_dm[i] remains 0.0

    atr_arr = wilder_atr(high, low, close, length)

    plus_dm_rma = rma(plus_dm, length)
    minus_dm_rma = rma(minus_dm, length)

    plus_di = [math.nan] * n
    minus_di = [math.nan] * n

    for i in range(n):
        atr_val = atr_arr[i] # Can be NaN
        # Division by zero or NaN atr_val
        if not math.isnan(atr_val) and atr_val != 0:
            # plus_dm_rma[i] can be NaN
            if not math.isnan(plus_dm_rma[i]):
                plus_di[i] = (100.0 * plus_dm_rma[i]) / atr_val
            if not math.isnan(minus_dm_rma[i]):
                minus_di[i] = (100.0 * minus_dm_rma[i]) / atr_val
        # else DI remains NaN

    dx = [math.nan] * n
    for i in range(n):
        pdi = plus_di[i]
        mdi = minus_di[i]
        if not math.isnan(pdi) and not math.isnan(mdi):
            sum_di = pdi + mdi
            if sum_di != 0: # Avoid division by zero
                dx[i] = (100.0 * abs(pdi - mdi)) / sum_di
            # else dx[i] remains NaN (covers pdi+mdi=0, leading to NaN in TS due to X/0 or 0/0)

    adx = rma(dx, length)

    return (plus_di, minus_di, adx)

# /* ───────── session VWAP (Resets each calendar day) ───────── */
def vwap_session(
    close: List[float], volume: List[float], timestamp: List[int]
) -> List[float]:
    """Calculates session-based VWAP, resetting daily."""
    n = len(close)
    if n == 0 or not (n == len(volume) == len(timestamp)):
        return [math.nan] * n if n > 0 else []

    out = [math.nan] * n

    def to_ms_ts(t: int) -> int: # Ensure timestamp is in milliseconds
        return t * 1000 if t < 1_000_000_000_000 else t

    sum_pv = 0.0
    sum_v = 0.0

    # JS toDateString() is locale-specific for its string format but represents a specific day.
    # For Python, to match, use local timezone from timestamp for date boundary.
    # A fixed format like YYYY-MM-DD is generally stabler.
    # datetime.fromtimestamp(seconds_since_epoch) uses local timezone by default.
    try:
        # Initial day string based on local timezone interpretation of timestamp
        first_ts_ms = to_ms_ts(timestamp[0])
        cur_day_str = datetime.fromtimestamp(first_ts_ms / 1000.0).strftime('%Y-%m-%d')
    except IndexError: # Should be caught by n==0
        return []

    for i in range(n):
        current_close = close[i]
        current_volume = volume[i]
        ts_ms = to_ms_ts(timestamp[i])

        # NaN propagation: if current_close or current_volume is NaN, sum_pv/sum_v become NaN

        day_str_loop = datetime.fromtimestamp(ts_ms / 1000.0).strftime('%Y-%m-%d')

        if day_str_loop != cur_day_str: # New day
            sum_pv = 0.0
            sum_v = 0.0
            cur_day_str = day_str_loop

        # If current_close or current_volume is NaN, product is NaN. sum_pv becomes NaN.
        sum_pv += current_close * current_volume
        # If current_volume is NaN, sum_v becomes NaN.
        sum_v += current_volume

        # Check for NaN in sums before division
        if math.isnan(sum_pv) or math.isnan(sum_v):
            out[i] = math.nan
        elif sum_v != 0:
            out[i] = sum_pv / sum_v
        else: # sum_v is 0 (and not NaN)
            out[i] = current_close # Fallback to current close price

    return out

# /* ───────── bullish-probability ───────── */
def bullish_probability(
    rsi: float, macd_hist: float, adx: float, st_k: float, st_d: float,
    price: float, vwap_val: float,
    lips: float, teeth: float, jaw: float
) -> float:
    """Calculates a bullish probability score."""
    count = 0
    # as_bool handles None/NaN correctly for conditions
    count += 1 if as_bool(rsi > 50) else 0
    count += 1 if as_bool(macd_hist > 0) else 0
    count += 1 if as_bool(adx > 25) else 0
    count += 1 if as_bool(st_k > st_d and st_k > 50) else 0
    count += 1 if as_bool(price > vwap_val) else 0
    count += 1 if as_bool(lips > teeth and teeth > jaw) else 0

    probability = (count / 6.0) * 100.0
    # Emulate Number(...toFixed(2)): convert to string with 2 decimal places, then to float
    # This also handles rounding like toFixed (0.5 rounds away from zero).
    # Python's f-string formatting with .2f rounds .5 to nearest even.
    # For precise toFixed(2) behavior:
    if math.isnan(probability): return math.nan
    return float(f"{probability:.2f}") # Standard rounding often used in Python.
                                        # For exact JS .toFixed() rounding:
                                        # temp_str = format(Decimal(str(probability)), '.2f') # using Decimal for precise rounding
                                        # return float(temp_str)
                                        # Or simpler if precision needs are met by f-string:
                                        # return round(probability * 100) / 100 # Not quite toFixed
                                        # The provided TS likely relies on standard float to string formatting.

# /* ───────── probability label ───────── */
def _custom_round_js_style(val: float) -> int:
    """Emulates JavaScript's Math.round (0.5 rounds away from zero)."""
    if math.isnan(val): return 0 # Or handle as error/NaN string
    if val >= 0:
        return math.floor(val + 0.5)
    else:
        return math.ceil(val - 0.5)

def probability_label(p: float) -> str:
    """Generates a descriptive label based on probability."""
    desc = ""
    if math.isnan(p):
        desc = "Unknown"
    elif p == 0:
        desc = 'Sideways'
    elif p <= 30:
        desc = 'Bearish'
    elif p <= 40:
        desc = 'Koreksi Lanjutan'
    elif p <= 50:
        desc = 'Konsolidasi'
    elif p <= 60:
        desc = 'Teknikal Rebound'
    else: # p > 60
        desc = 'Probabilitas Bullish'

    rounded_p_str = str(_custom_round_js_style(p)) if not math.isnan(p) else "N/A"
    return f"{desc} ({rounded_p_str}%)"

# /* ───────── stage detector ───────── */
def stage_name(
    close_val: float, macd_l_now: float, macd_l_prev: float,
    macd_s_now: float, macd_s_prev: float,
    rsi_val: float, ma50_val: float
) -> str:
    """Detects market stage based on indicators."""
    # NaN comparisons evaluate to False, naturally leading to 'Netral' if critical values are NaN.
    cond1 = (macd_l_prev < macd_s_prev and macd_l_now > macd_s_now and
             rsi_val > 40 and rsi_val < 60 and
             close_val < ma50_val)
    if as_bool(cond1): return '1: Akumulasi' # Using as_bool for safety with potential None/NaN inputs

    cond2 = (macd_l_now > macd_s_now and
             rsi_val > 55 and
             close_val > ma50_val)
    if as_bool(cond2): return '2: Tren Naik'

    cond3 = (macd_l_prev > macd_s_prev and macd_l_now < macd_s_now and
             rsi_val > 60 and rsi_val < 70)
    if as_bool(cond3): return '3: Distribusi'

    cond4 = (macd_l_now < macd_s_now and
             rsi_val < 45 and
             close_val < ma50_val)
    if as_bool(cond4): return '4: Tren Turun'

    return 'Netral'

# Helper for arfoxScoreSeries: pandas-like shift
def _shift_series(series: List[float], periods: int) -> List[float]:
    n = len(series)
    if periods == 0:
        return list(series) # Return a copy

    shifted = [math.nan] * n
    if periods > 0: # Positive shift, values from the past: shifted[i] = series[i-periods]
        for i in range(periods, n):
            shifted[i] = series[i - periods]
    else: # Negative shift (not used in TS), values from the future
        abs_periods = abs(periods)
        for i in range(n - abs_periods):
            shifted[i] = series[i + abs_periods]
    return shifted

# /* ───────── full Arfox raw-score series ───────── */
def arfox_score_series(
    price: List[float], volume: List[float], high: List[float], low: List[float], timestamp_ms: List[int]
) -> List[float]:
    """Calculates the Arfox raw score series."""
    n_periods = len(price)
    if n_periods == 0: return []

    ma_local = rolling_mean # Use the globally defined rolling_mean

    ma5 = ma_local(price, 5)
    ma20 = ma_local(price, 20)
    ma50 = ma_local(price, 50)
    ma100 = ma_local(price, 100)
    ma200 = ma_local(price, 200)
    ma10v = ma_local(volume, 10)

    prev_price = [math.nan] * n_periods
    prev_vol = [math.nan] * n_periods
    if n_periods > 0:
        prev_price[0] = price[0] # TS: [price[0]].concat(price.slice(0,-1)) -> prevPrice[0] = price[0]
        prev_vol[0] = volume[0]   # Same for volume
        for i in range(1, n_periods):
            prev_price[i] = price[i-1]
            prev_vol[i] = volume[i-1]

    _macd_l, _macd_s, macd_hist = macd_calc(price)
    _plus_di, _minus_di, adx_arr = dmi_calc(high, low, price)
    st_k_arr, st_d_arr = stoch_kd(price, high, low)

    high_roll_max10 = rolling_max(high, 10)
    low_roll_min10 = rolling_min(low, 10)
    rng10 = [(hr - lr) if not math.isnan(hr) and not math.isnan(lr) else math.nan
             for hr, lr in zip(high_roll_max10, low_roll_min10)]

    std20 = rolling_std(price, 20)
    bbw = [(s * 2.0) if not math.isnan(s) else math.nan for s in std20]
    bbw50 = ma_local(bbw, 50)

    obv = [0.0] * n_periods
    if n_periods > 0:
        acc_obv = 0.0
        # obv[0] = 0 as sign for i=0 is 0 in TS logic
        for i in range(n_periods):
            sign_val = 0.0
            if i > 0:
                price_diff = price[i] - price[i-1]
                if math.isnan(price_diff): sign_val = math.nan # Match JS Math.sign(NaN) = NaN
                elif price_diff > 0: sign_val = 1.0
                elif price_diff < 0: sign_val = -1.0
                # else sign_val is 0.0

            term = sign_val * volume[i] # This can be NaN if sign_val or volume[i] is NaN

            if math.isnan(acc_obv): pass # acc_obv remains NaN
            elif math.isnan(term): acc_obv = math.nan
            else: acc_obv += term
            obv[i] = acc_obv
    obv50 = ma_local(obv, 50)

    vwap_arr = vwap_session(price, volume, timestamp_ms)
    atr14 = wilder_atr(high, low, price, 14)
    atr50 = ma_local(atr14, 50)

    # Alligator lines using shifted MAs
    lips = _shift_series(ma_local(price, 5), 3)
    teeth = _shift_series(ma_local(price, 8), 5)
    jaw = _shift_series(ma_local(price, 13), 8)

    score = [10.0] * n_periods

    # Use the globally defined wilder_rsi
    rsi_arr_for_score = wilder_rsi(price, 14)

    def add_score_item(idx: int, condition_val: bool, points_if_true: float, points_if_false: float):
        # condition_val is already a resolved boolean from Python's NaN comparison behavior.
        score[idx] += points_if_true if condition_val else points_if_false

    for i in range(n_periods):
        # Explicit NaN checks for conditions to ensure safety and clarity
        p_i, ma5_i, pp_i = price[i], ma5[i], prev_price[i]
        v_i, ma10v_i, pv_i = volume[i], ma10v[i], prev_vol[i]
        ma20_i, ma50_i = ma20[i], ma50[i]
        ma100_i, ma200_i = ma100[i], ma200[i]
        rsi_i, macd_h_i, adx_i_sc = rsi_arr_for_score[i], macd_hist[i], adx_arr[i] # Renamed adx_i to adx_i_sc
        rng10_i, stk_i, std_i = rng10[i], st_k_arr[i], st_d_arr[i]
        bbw_i, bbw50_i_sc = bbw[i], bbw50[i] # Renamed bbw50_i to bbw50_i_sc
        obv_i, obv50_i_sc = obv[i], obv50[i] # Renamed obv50_i to obv50_i_sc
        vwap_i, atr14_i, atr50_i_sc = vwap_arr[i], atr14[i], atr50[i] # Renamed atr50_i to atr50_i_sc
        lips_i, teeth_i, jaw_i = lips[i], teeth[i], jaw[i]

        add_score_item(i, not math.isnan(p_i) and p_i >= 60, 10, -5)
        add_score_item(i, not math.isnan(p_i) and not math.isnan(ma5_i) and p_i >= ma5_i, 10, -5)
        add_score_item(i, not math.isnan(p_i) and not math.isnan(pp_i) and p_i > pp_i, 10, -5)
        add_score_item(i, not math.isnan(pp_i) and pp_i >= 1, 5, -5)

        change_cond = False
        if not math.isnan(p_i) and not math.isnan(pp_i) and pp_i != 0:
            change = ((p_i - pp_i) / pp_i) * 100.0
            if not math.isnan(change) and change > 1: change_cond = True
        add_score_item(i, change_cond, 10, -5)

        vol_cond1 = False
        if not math.isnan(v_i) and not math.isnan(ma10v_i) and ma10v_i != 0 : # Check ma10v_i != 0 if it could be
             if v_i >= 2 * ma10v_i : vol_cond1 = True
        elif not math.isnan(v_i) and not math.isnan(ma10v_i) and ma10v_i == 0 and v_i >=0 : # v_i >= 2*0
             vol_cond1 = True
        add_score_item(i, vol_cond1, 10, -5)

        add_score_item(i, not math.isnan(v_i) and not math.isnan(pv_i) and v_i >= pv_i, 10, -5)

        turnover_cond = False
        if not math.isnan(v_i) and not math.isnan(p_i):
            if (v_i * p_i) >= 5e10: turnover_cond = True
        add_score_item(i, turnover_cond, 10, -10)

        score[i] += 5 # bandar placeholder

        cross_up, cross_dn = False, False
        if i > 0: # Need previous values for MAs
            ma20_prev, ma50_prev = ma20[i-1], ma50[i-1]
            if not math.isnan(ma20_prev) and not math.isnan(ma50_prev) and \
               not math.isnan(ma20_i) and not math.isnan(ma50_i):
                if ma20_prev < ma50_prev and ma20_i > ma50_i: cross_up = True
                if ma20_prev > ma50_prev and ma20_i < ma50_i: cross_dn = True
        add_score_item(i, cross_up, 20, 0)
        add_score_item(i, cross_dn, -20, 0) # if true, add -20, else add 0.

        add_score_item(i, not math.isnan(ma20_i) and not math.isnan(ma50_i) and ma20_i > ma50_i, 15, -10)
        add_score_item(i, not math.isnan(ma50_i) and not math.isnan(ma100_i) and ma50_i > ma100_i, 15, -10)
        add_score_item(i, not math.isnan(ma100_i) and not math.isnan(ma200_i) and ma100_i > ma200_i, 15, -10)

        add_score_item(i, not math.isnan(rsi_i) and rsi_i > 50, 5, -5)
        add_score_item(i, not math.isnan(macd_h_i) and macd_h_i > 0, 5, -5)
        add_score_item(i, not math.isnan(adx_i_sc) and adx_i_sc > 25, 10, -5)

        rng_contr_cond = False
        if not math.isnan(rng10_i) and not math.isnan(p_i) and p_i != 0:
            if rng10_i < (p_i * 0.02): rng_contr_cond = True
        elif not math.isnan(rng10_i) and not math.isnan(p_i) and p_i == 0 and rng10_i < 0: # rng10_i < 0 if p_i is 0
             rng_contr_cond = True # If price is 0, 2% of price is 0. Range must be < 0 (e.g. negative range, not typical)
        add_score_item(i, rng_contr_cond, -5, 0)

        stoch_bull_cond = False
        if not math.isnan(stk_i) and not math.isnan(std_i):
            if stk_i > std_i and stk_i > 50: stoch_bull_cond = True
        add_score_item(i, stoch_bull_cond, 5, -5)

        add_score_item(i, not math.isnan(bbw_i) and not math.isnan(bbw50_i_sc) and bbw_i > bbw50_i_sc, 5, 0)
        add_score_item(i, not math.isnan(obv_i) and not math.isnan(obv50_i_sc) and obv_i > obv50_i_sc, 5, 0)
        add_score_item(i, not math.isnan(p_i) and not math.isnan(vwap_i) and p_i > vwap_i, 5, -5)
        add_score_item(i, not math.isnan(atr14_i) and not math.isnan(atr50_i_sc) and atr14_i > atr50_i_sc, 5, 0)

        alligator_bull_cond = False
        if not math.isnan(lips_i) and not math.isnan(teeth_i) and not math.isnan(jaw_i):
            if lips_i > teeth_i and teeth_i > jaw_i: alligator_bull_cond = True
        add_score_item(i, alligator_bull_cond, 10, -10)

        current_score_val = score[i]
        if math.isnan(current_score_val): score[i] = 10.0 # Default to min if NaN
        else: score[i] = max(10.0, min(100.0, current_score_val))

    return score

# /* ───────── Conservative S/R ATR ───────── */
def sr_atr_conservative(
    high: List[float], low: List[float], atr_arr: List[float],
    sr_len: int = 20, atr_mult: float = 1.5
) -> Tuple[List[float], List[float], List[float], List[float]]:
    """Calculates conservative Support/Resistance levels using ATR."""
    n = len(high)
    if not (n == len(low) == len(atr_arr)):
        if n > 0: # Base length on high if available
            nan_list = [math.nan] * n
            return (nan_list, nan_list, nan_list, nan_list)
        return ([], [], [], []) # All inputs potentially empty

    support = rolling_min(low, sr_len)
    resistance = rolling_max(high, sr_len)

    sl_con = [(s - atr_arr[i] * atr_mult) if not math.isnan(s) and i < len(atr_arr) and not math.isnan(atr_arr[i]) else math.nan
              for i, s in enumerate(support)]

    tp_con = [(r + atr_arr[i] * atr_mult) if not math.isnan(r) and i < len(atr_arr) and not math.isnan(atr_arr[i]) else math.nan
              for i, r in enumerate(resistance)]

    return (support, resistance, sl_con, tp_con)

# Define a type hint for the candle data for clarity
Candle = Dict[str, Any]

def fetch_yahoo(
    symbol: str,
    interval: str = '1h',
    start_date: str = None,
    end_date: str = None,
    max_retry: int = 3,
    timeout: int = 15
) -> List[Candle]:
    """
    Fetches historical market data from Yahoo Finance with retry and timeout logic.
    """
    start_ts = int(datetime.strptime(start_date, '%Y-%m-%d').timestamp())
    end_ts = int(datetime.strptime(end_date, '%Y-%m-%d').timestamp())

    api_url = (
        f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}"
        f"?period1={start_ts}&period2={end_ts}&interval={interval}"
        f"&includePrePost=true&events=div|split"
    )
    print(api_url)
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
    }

    res = None
    for attempt in range(1, max_retry + 1):
        try:
            res = requests.get(api_url, headers=headers, timeout=timeout)
            res.raise_for_status()
            break
        except (requests.exceptions.RequestException, requests.exceptions.HTTPError) as e:
            # print(f"Attempt {attempt} for {symbol} failed: {e}")
            if attempt == max_retry:
                return [] # Return empty list on failure
            time.sleep(1 * attempt)

    if not res:
        return []

    js = res.json()
    chart_result = js.get('chart', {}).get('result')
    if not chart_result or not chart_result[0]:
        return []

    res_data = chart_result[0]
    timestamps = res_data.get('timestamp', [])
    quote = res_data.get('indicators', {}).get('quote', [{}])[0]

    candles: List[Candle] = []
    for i, t in enumerate(timestamps):
        candles.append({
            't': t * 1000,
            'o': quote.get('open', [])[i], 'h': quote.get('high', [])[i],
            'l': quote.get('low', [])[i], 'c': quote.get('close', [])[i],
            'v': quote.get('volume', [])[i],
        })

    return [c for c in candles if c.get('c') is not None]

Row = Dict[str, Any]

# IDX tick size helpers
def tick_step(p: float) -> int:
    if p < 200: return 1
    if p < 500: return 2
    if p < 2000: return 5
    if p < 5000: return 10
    return 25

def round_idx(p: float, direction: str = 'nearest') -> int:
    if math.isnan(p): return p
    s = tick_step(p)
    if direction == 'up': return math.ceil(p / s) * s
    if direction == 'down': return math.floor(p / s) * s
    return round(p / s) * s

# Price formatter
def fmt(p: float, mkt: str, direction: str = 'nearest') -> str:
    if math.isnan(p): return 'N/A'
    if mkt == 'IDX':
        return str(round_idx(p, direction))

    d = 2 if p >= 1 else 4 # For US Market
    if mkt == 'CRYPTO': d = 4
    return f"{p:.{d}f}"

# Trend flip helper
def flip_since(trend: List[int], look: int = 60) -> Dict[str, int]:
    if not trend: return {'bars': 0}
    cur = last(trend)
    i = len(trend) - 1
    while i > 0 and trend[i] == cur and (len(trend) - 1 - i) < look:
        i -= 1
    idx = i + 1
    return {'bars': len(trend) - 1 - idx}


def create_features_for_df(df: pd.DataFrame, timeframe_label: str) -> Dict[str, float]:
    """
    Calculates a comprehensive and extensive set of features for a given dataframe
    and returns the last value of each.
    """
    if df.empty or len(df) < 250:
        return {}

    features = {}
    # Extract lists from the dataframe
    open_p = df['open'].tolist()
    close = df['close'].tolist()
    high = df['high'].tolist()
    low = df['low'].tolist()
    volume = df['volume'].tolist()
    timestamps_ms = (df.index.astype(np.int64) // 10**6).tolist()
    last_close = last(close)

    # --- Foundational Indicators (used by other features) ---
    atr14 = wilder_atr(high, low, close, 14)
    last_atr14 = last(atr14)

    ## From build_row: ema50 is needed for trend_bull used in rsi_label ##
    ema50 = ema(close, 50)
    last_ema50 = last(ema50)
    trend_bull = last_close > last_ema50 if not math.isnan(last_close) and not math.isnan(last_ema50) else False

    # --- 1. Price & Moving Average Features ---
    sma8 = rolling_mean(close, 8)
    sma20 = rolling_mean(close, 20)
    sma50 = rolling_mean(close, 50)
    sma200 = rolling_mean(close, 200)
    features['price_vs_sma20'] = (last_close / last(sma20)) - 1 if last(sma20) and not math.isnan(last(sma20)) else np.nan
    features['price_vs_sma50'] = (last_close / last(sma50)) - 1 if last(sma50) and not math.isnan(last(sma50)) else np.nan
    features['sma20_vs_sma50'] = (last(sma20) / last(sma50)) - 1 if last(sma50) and not math.isnan(last(sma50)) else np.nan
    features['sma50_vs_sma200'] = (last(sma50) / last(sma200)) - 1 if last(sma200) and not math.isnan(last(sma200)) else np.nan
    # Inspired by ma_labels: numerical representation of MA stack
    if last(sma8) > last(sma20) > last(sma50): features['ma_stack'] = 1
    elif last(sma8) < last(sma20) < last(sma50): features['ma_stack'] = -1
    else: features['ma_stack'] = 0

    # --- 2. Momentum & Trend Features ---
    features['rsi_14'] = last(wilder_rsi(close, 14))
    macdL, macdS, macd_hist = macd_calc(close)
    features['macd_hist'] = last(macd_hist)
    stoch_k, stoch_d = stoch_kd(close, high, low, 14)
    features['stoch_k'] = last(stoch_k)
    features['stoch_d'] = last(stoch_d)
    plus_di, minus_di, adx = dmi_calc(high, low, close, 14)
    features['adx_14'] = last(adx)
    features['dmi_diff'] = last(plus_di) - last(minus_di)
    # Rate of Change (ROC) for 10 periods
    if len(close) > 10: features['roc_10'] = (last_close / close[-11] - 1) if close[-11] != 0 else np.nan
    
    # Inspired by build_row: SuperTrend features
    st_line, st_trend = super_trend(close, high, low)
    flip_info = flip_since(st_trend)
    idx_start = len(st_trend) - 1 - flip_info['bars']
    entry_px = st_line[idx_start - 1] if idx_start > 0 else st_line[idx_start]
    features['supertrend_dir'] = last(st_trend)
    features['price_vs_supertrend'] = (last_close / last(st_line)) - 1 if last(st_line) else np.nan
    features['bars_since_st_flip'] = flip_info['bars']
    features['pl_since_st_flip'] = (last_close / entry_px - 1) if entry_px and not math.isnan(entry_px) else np.nan

    # --- 3. Volatility Features ---
    features['atr_14_norm'] = (last_atr14 / last_close) if last_close and not math.isnan(last_close) else np.nan
    # Bollinger Bands
    std20 = rolling_std(close, 20)
    bb_mid = sma20
    bb_upper = [m + 2 * s for m, s in zip(bb_mid, std20)]
    bb_lower = [m - 2 * s for m, s in zip(bb_mid, std20)]
    bb_width = [(u - l) / m if m and not math.isnan(m) else np.nan for u, l, m in zip(bb_upper, bb_lower, bb_mid)]
    bb_percent_b = [(last_close - l) / (u - l) if (u-l) != 0 else np.nan for u,l in [(last(bb_upper), last(bb_lower))]]
    features['bb_width'] = last(bb_width)
    features['bb_percent_b'] = last(bb_percent_b)
    
    # Inspired by build_row: Williams VIX Fix
    wvf, wvf_upper, _ = foxpro_wvf(close, low)
    features['wvf_raw'] = wvf
    features['wvf_vs_upper'] = (wvf / wvf_upper) - 1 if wvf_upper and not math.isnan(wvf_upper) else np.nan

    # --- 4. Volume & High-Level Features ---
    vwap = vwap_session(close, volume, timestamps_ms)
    features['price_vs_vwap'] = (last_close / last(vwap)) - 1 if last(vwap) and not math.isnan(last(vwap)) else np.nan
    vol_sma20 = rolling_mean(volume, 20)
    features['volume_vs_sma20'] = (last(volume) / last(vol_sma20)) - 1 if last(vol_sma20) and not math.isnan(last(vol_sma20)) else np.nan
    # Inspired by build_row: Arfox Score
    score_series = arfox_score_series(close, volume, high, low, timestamps_ms)
    features['arfox_score'] = last(score_series)
    features['arfox_score_ma20'] = last(rolling_mean(score_series, 20))
    # Inspired by build_row: Stage Analysis (numerical)
    stage_str = stage_name(last_close, last(macdL), macdL[-2], last(macdS), macdS[-2], features['rsi_14'], last(sma50))
    stage_map = {'1: Akumulasi': 1, '2: Tren Naik': 2, '3: Distribusi': 3, '4: Tren Turun': 4}
    features['market_stage'] = stage_map.get(stage_str, 0) # 0 for Neutral

    ## From build_row: Bullish Probability ##
    lips, teeth, jaw = last(_shift_series(rolling_mean(close, 5), 3)), last(_shift_series(rolling_mean(close, 8), 5)), last(_shift_series(rolling_mean(close, 13), 8))
    features['bullish_prob_score'] = bullish_probability(features['rsi_14'], last(macd_hist), features['adx_14'], features['stoch_k'], features['stoch_d'], last_close, last(vwap), lips, teeth, jaw)

    ## From build_row: Conservative S/R ##
    sup, res, sl_con, tp_con = sr_atr_conservative(high, low, atr14)
    features['price_vs_support'] = (last_close / last(sup) - 1) if last(sup) else np.nan
    features['price_vs_resistance'] = (last_close / last(res) - 1) if last(res) else np.nan
    features['price_vs_sl_conserve'] = (last_close / last(sl_con) - 1) if last(sl_con) else np.nan

    # --- 5. Price Action / Candlestick Features ---
    last_open = last(open_p)
    last_high = last(high)
    last_low = last(low)
    candle_range = last_high - last_low
    # Position of close within the full H-L range
    features['close_pos_in_range'] = (last_close - last_low) / candle_range if candle_range > 0 else 0.5
    # Normalized candle sizes
    if last_atr14 > 0:
        features['body_size_norm'] = abs(last_close - last_open) / last_atr14
        features['upper_wick_norm'] = (last_high - max(last_open, last_close)) / last_atr14
        features['lower_wick_norm'] = (min(last_open, last_close) - last_low) / last_atr14

    # --- 6. NEW: Volume Profile Features (Optimized) ---
    vp_df = df.iloc[-100:].copy()
    # Initialize features to NaN to handle cases where calculation is skipped
    features['volume_profile_hvn_dist'] = np.nan
    features['volume_profile_lvn_dist'] = np.nan
    features['volume_profile_va_ratio'] = np.nan

    if not vp_df.empty and vp_df['high'].max() > vp_df['low'].min():
        # Calculate Volume Profile
        price_range = vp_df['high'].max() - vp_df['low'].min()
        tick = tick_step(last_close)
        num_bins = int(price_range / tick) if tick > 0 else 20
        if num_bins < 2:
            num_bins = 2
        # Use observed=False to maintain old behavior and silence warning
        vp = vp_df.groupby(pd.cut(vp_df['close'], bins=num_bins, right=False), observed=False)['volume'].sum()

        # Find Point of Control (POC), HVNs, and LVNs
        if not vp.empty:
            volume_threshold = vp.mean()
            hvns = vp[vp > volume_threshold]
            lvns = vp[vp < volume_threshold]

            # Find nearest HVN and LVN
            if not hvns.empty:
                hvn_mids = pd.IntervalIndex(hvns.index).mid
                nearest_hvn = hvn_mids[np.abs(hvn_mids - last_close).argmin()]
                features['volume_profile_hvn_dist'] = (last_close / nearest_hvn - 1) if nearest_hvn != 0 else np.nan
            
            if not lvns.empty:
                lvn_mids = pd.IntervalIndex(lvns.index).mid
                nearest_lvn = lvn_mids[np.abs(lvn_mids - last_close).argmin()]
                features['volume_profile_lvn_dist'] = (last_close / nearest_lvn - 1) if nearest_lvn != 0 else np.nan

        # --- OPTIMIZED VALUE AREA CALCULATION ---
        total_volume = vp.sum()
        if total_volume > 0 and not vp.empty:
            # Sort bins by volume in descending order
            vp_sorted = vp.sort_values(ascending=False)
            
            # Calculate cumulative share of volume
            vp_cumsum_share = vp_sorted.cumsum() / total_volume
            
            # Filter to get the bins that make up the Value Area (70% of volume)
            value_area_bins = vp_sorted[vp_cumsum_share <= 0.70]
            
            if not value_area_bins.empty:
                # Get the min and max price intervals from this group
                va_intervals = pd.IntervalIndex(value_area_bins.index)
                va_low = va_intervals.left.min()
                va_high = va_intervals.right.max()

                # Calculate VA Ratio
                va_range = va_high - va_low
                if va_range > 0:
                    if last_close > va_high:
                        features['volume_profile_va_ratio'] = 1 + (last_close - va_high) / va_range
                    elif last_close < va_low:
                        features['volume_profile_va_ratio'] = 1 - (va_low - last_close) / va_range
                    else:
                        features['volume_profile_va_ratio'] = 1
                else: # Handle zero range case
                    features['volume_profile_va_ratio'] = 1 if last_close == va_low else (2 if last_close > va_high else 0)
    return features

def generate_data_for_timeframe(timeframe: str, tickers: List[str], cfg: Dict) -> pd.DataFrame:
    """
    Generates a complete training dataset for a single specified timeframe.
    It fetches data once per ticker, then samples and processes it.
    """
    all_data_rows = []
    target_horizons = cfg["TARGET_HORIZONS"].get(timeframe, {})
    if not target_horizons:
        print(f"Warning: No target horizons defined for timeframe {timeframe}. Skipping.")
        return pd.DataFrame()

    for ticker in tqdm(tickers, desc=f"Processing Tickers for {timeframe}"):
        # 1. Fetch one large chunk of data for the ticker for this timeframe
        fetch_start_dt = datetime.strptime(cfg["DATA_START_DATE"], '%Y-%m-%d') - timedelta(days=cfg["HISTORY_BUFFER_DAYS"])
        master_candles = fetch_yahoo(
            symbol=ticker,
            interval=timeframe,
            start_date=fetch_start_dt.strftime('%Y-%m-%d'),
            end_date=cfg["DATA_END_DATE"]
        )
        master_df = candles_to_dataframe(master_candles)
        if master_df.empty:
            print(f"DEBUG: fetch_yahoo returned no data for {ticker} on timeframe {timeframe}. Skipping.")
            continue

        # 2. FIX: Identify a valid window for sampling that guarantees enough history for feature creation.
        min_history_required = 250 # As defined in create_features_for_df

        # Find the first possible date we can sample from.
        first_valid_index_date = master_df.index[min_history_required] if len(master_df) > min_history_required else None

        # If there's no valid date (not enough data overall), skip this ticker.
        if first_valid_index_date is None:
            print(f"DEBUG: {ticker} has fewer than {min_history_required} total data points. Skipping.")
            continue

        # --- END BUFFER: Find the last possible date we can sample from ---
        max_horizon_candles = max(target_horizons.values()) if target_horizons else 0
        last_valid_index_date = master_df.index[-max_horizon_candles -1] if len(master_df) > max_horizon_candles else None

        if last_valid_index_date is None:
            print(f"DEBUG: {ticker} does not have enough future data for the longest target horizon. Skipping.")
            continue

        # --- Define the final sampling window with both buffers applied ---
        sampling_start_date = max(pd.to_datetime(cfg["DATA_START_DATE"]), first_valid_index_date)
        sampling_end_date = min(pd.to_datetime(cfg["DATA_END_DATE"]), last_valid_index_date)

        sampling_window_df = master_df[
          (master_df.index >= sampling_start_date) &
          (master_df.index < sampling_end_date)
        ]
        if sampling_window_df.empty:
              print(f"DEBUG: No data for {ticker} in the adjusted sampling window. Skipping.")
              continue

        # 3. Get evenly spaced timestamps instead of random ones.
        n_samples = cfg["ROWS_PER_STOCK"]
        total_available_points = len(sampling_window_df)

        if total_available_points < n_samples:
           # If we don't have enough data points for the desired sample size, use all available points.
           valid_timestamps = sampling_window_df.index.tolist()
        else:
           # Use np.linspace to get N evenly spaced indices from the start to the end of the dataframe.
           indices = np.linspace(0, total_available_points - 1, num=n_samples, dtype=int)
           print(total_available_points/n_samples)
           valid_timestamps = sampling_window_df.iloc[indices].index.tolist()

        # 3. For each sampled timestamp, generate features and targets
        for ts in tqdm(valid_timestamps, desc=f"Sampling {ticker}", leave=False):
            # --- Feature Generation ---
            historical_df = master_df[master_df.index <= ts]
            feature_set = create_features_for_df(historical_df, timeframe)
            if not feature_set:
              print(f"DEBUG: Feature creation failed for {ticker} at {ts}. History length: {len(historical_df)}")
              continue

            feature_set['ticker'] = ticker
            feature_set['timestamp'] = ts

            # --- Target Calculation ---
            future_df = master_df[master_df.index > ts]
            current_price = historical_df.iloc[-1]['close']

            if np.isnan(current_price) or current_price == 0:
                continue

            for name, horizon_candles in target_horizons.items():
                if len(future_df) >= horizon_candles:
                    future_candle = future_df.iloc[horizon_candles - 1]
                    future_price = future_candle['close']
                    pct_change = (future_price - current_price) / current_price

                    feature_set[f"{name}_pct_change"] = pct_change
                    feature_set[f"{name}_end_time"] = future_candle.name
                else:
                    feature_set[f"{name}_pct_change"] = np.nan
                    feature_set[f"{name}_end_time"] = pd.NaT
                
                # # --- NEW: Triple Barrier Label Calculation ---
                # label = 0 # Default to 0 (Hold/Timeout)
                # barrier_config = cfg.get("TRIPLE_BARRIER_CONFIG", {}).get(name)

                # if barrier_config and len(future_df) >= horizon_candles:
                #     upper_barrier = current_price * (1 + barrier_config["up"])
                #     lower_barrier = current_price * (1 + barrier_config["down"])

                #     # Look at the price path over the defined horizon
                #     path = future_df.iloc[:horizon_candles]

                #     for _, candle in path.iterrows():
                #         if candle['high'] >= upper_barrier:
                #             label = 1  # Price hit take-profit first
                #             break
                #         if candle['low'] <= lower_barrier:
                #             label = -1 # Price hit stop-loss first
                #             break
                # else:
                #     label = np.nan # Not enough data to determine label

                # feature_set[f"{name}_label"] = label

                # --- NEW: Enhanced Triple Barrier (Level 1) ---
                # 2: Strong Buy, 1: Weak Buy (Fakeout), 0: Hold, -1: Weak Sell (Fakeout), -2: Strong Sell
                label = 0 # Default to Hold/Timeout
                barrier_config = cfg.get("TRIPLE_BARRIER_CONFIG", {}).get(name)
 
                if barrier_config and len(future_df) >= horizon_candles:
                   upper_barrier = current_price * (1 + barrier_config["up"])
                   lower_barrier = current_price * (1 + barrier_config["down"])
                   path = future_df.iloc[:horizon_candles]

                   for i, candle in enumerate(path.itertuples()):
                       # Check for upper barrier touch
                       if candle.high >= upper_barrier:
                           label = 2 # Provisionally a Strong Buy
                           # Check rest of path for a reversal to the lower barrier
                           remaining_path = path.iloc[i+1:]
                           if not remaining_path.empty and (remaining_path['low'] <= lower_barrier).any():
                               label = 1 # It's a Weak Buy (bull trap)
                           break # Outcome determined

                       # Check for lower barrier touch
                       if candle.low <= lower_barrier:
                           label = -2 # Provisionally a Strong Sell
                           # Check rest of path for a reversal to the upper barrier
                           remaining_path = path.iloc[i+1:]
                           if not remaining_path.empty and (remaining_path['high'] >= upper_barrier).any():
                               label = -1 # It's a Weak Sell (bear trap)
                           break # Outcome determined
                else:
                   label = np.nan # Not enough data to determine the label

                feature_set[f"{name}_label"] = label

            all_data_rows.append(feature_set)

    if not all_data_rows:
        return pd.DataFrame()

    # 4. Post-Processing: Convert to DataFrame and calculate final scores
    full_df = pd.DataFrame(all_data_rows)
    # DEBUG: Check the state of the DataFrame *before* dropping rows.
    # if full_df.empty:
        # print("DEBUG: No rows were generated after sampling. Check previous debug messages.")
        # return pd.DataFrame()
    # print(f"DEBUG: Generated {len(full_df)} rows before dropping NaNs. Checking rsi_14...")
    # print(full_df[['ticker', 'rsi_14']].to_string())
    # full_df.dropna(subset=['rsi_14'], inplace=True) # Ensure key features are present

    print("\nCalculating benchmarks and final scores...")
    fixed_benchmarks = cfg.get("FIXED_BENCHMARKS", {})
    for name in tqdm(target_horizons.keys(), desc="Scoring Targets"):
        pct_change_col = f"{name}_pct_change"
        if pct_change_col not in full_df.columns:
            continue

        # Calculate and store the benchmark (for debugging)
        benchmark = fixed_benchmarks.get(name)
        # If no fixed benchmark is defined for this target name, skip scoring it.
        if benchmark is None:
            print(f"Warning: No fixed benchmark found for '{name}'. Skipping scoring for this target.")
            continue

        # full_df[f"{name}_avg_benchmark_change"] = benchmark

        # Calculate the final score
        if benchmark == 0 or np.isnan(benchmark):
            full_df[name] = 0.5
        else:
            ratio = full_df[pct_change_col].fillna(0) / benchmark
            score = 0.5 + (ratio * cfg["SCORE_SCALING_FACTOR"])
            full_df[name] = score.clip(0.0, 1.0)

    # 5. Final Formatting
    # Rename and format columns for final output
    jakarta_tz = 'Asia/Jakarta'
    full_df.rename(columns={'timestamp': 'start_time'}, inplace=True)
    full_df['start_time_gmt7'] = pd.to_datetime(full_df['start_time']).dt.tz_localize('UTC').dt.tz_convert(jakarta_tz).dt.strftime('%Y-%m-%d %H:%M:%S')

    for name in target_horizons.keys():
        # Format percentage change
        pct_col = f"{name}_pct_change"
        if pct_col in full_df.columns:
            full_df[pct_col] = full_df[pct_col].apply(lambda x: f"{x:+.2%}" if pd.notna(x) else "N/A")

        # Format end time
        end_time_col = f"{name}_end_time"
        if end_time_col in full_df.columns:
            new_end_time_col = f"{end_time_col}_gmt7"
            full_df[new_end_time_col] = pd.to_datetime(full_df[end_time_col]).dt.tz_localize('UTC').dt.tz_convert(jakarta_tz).dt.strftime('%Y-%m-%d %H:%M:%S')
            full_df.drop(columns=[end_time_col], inplace=True)

    # Reorder columns for readability
    id_cols = ['ticker', 'start_time_gmt7']

    # --- FIX: Identify feature columns by excluding known ID and target columns ---
    target_cols = sorted([c for c in full_df.columns if c.startswith('target')])
    known_non_feature_cols = set(id_cols + target_cols + ['start_time'])
    feature_cols = sorted([c for c in full_df.columns if c not in known_non_feature_cols])

    # Construct the final list of columns in the desired order
    final_cols = id_cols + feature_cols + target_cols
    return full_df[final_cols]


def candles_to_dataframe(candles: List[Dict[str, Any]]) -> pd.DataFrame:
    """Converts the List[Candle] from fetch_yahoo into a pandas DataFrame."""
    if not candles:
        return pd.DataFrame()
    df = pd.DataFrame(candles)
    df['timestamp'] = pd.to_datetime(df['t'], unit='ms')
    df.set_index('timestamp', inplace=True)
    df.rename(columns={'o': 'open', 'h': 'high', 'l': 'low', 'c': 'close', 'v': 'volume'}, inplace=True)
    df.drop(columns=['t'], inplace=True)
    # Ensure data types are correct, handling potential None values
    for col in ['open', 'high', 'low', 'close', 'volume']:
        df[col] = pd.to_numeric(df[col], errors='coerce')
    return df