from typing import List
import pandas as pd


def main(context, datas: List[pd.DataFrame]) -> pd.DataFrame:
    if not datas:
        return pd.DataFrame()
    df = datas[0]
    context.log.info(f"Received {len(df)} upstream rows", source="Custom data processing")
    return df

from typing import List
import pandas as pd


def main(context, datas: List[pd.DataFrame]) -> pd.DataFrame:
    df = datas[0].copy()

    # Calculate EMAs separately by symbol so instruments are never mixed
    df = df.sort_values(["symbol", "timestamp"])
    grouped = df.groupby("symbol", group_keys=False)
    df["ema_fast"] = grouped["close"].apply(lambda s: s.ewm(span=9, adjust=False).mean())
    df["ema_slow"] = grouped["close"].apply(lambda s: s.ewm(span=21, adjust=False).mean())

    # Bullish crossover=1 (long), bearish crossover=-1 (short), otherwise=0 (wait)
    diff = df["ema_fast"] - df["ema_slow"]
    prev = grouped["ema_fast"].shift(1) - grouped["ema_slow"].shift(1)
    df["prediction"] = 0
    df.loc[(diff > 0) & (prev <= 0), "prediction"] = 1
    df.loc[(diff < 0) & (prev >= 0), "prediction"] = -1

    context.log.info(
        f"Signals generated: {int((df['prediction'] == 1).sum())} long, "
        f"{int((df['prediction'] == -1).sum())} short",
        source="Custom data processing",
    )
    return df[["timestamp", "symbol", "close", "ema_fast", "ema_slow", "prediction"]]

from typing import List
import pandas as pd
from functools import reduce


def main(context, datas: List[pd.DataFrame]) -> pd.DataFrame:
    if not datas:
        return pd.DataFrame()

    # Outer-join every upstream DataFrame on timestamp + symbol
    merged = reduce(
        lambda left, right: pd.merge(left, right, on=["timestamp", "symbol"], how="outer"),
        datas,
    )
    merged = merged.sort_values(["symbol", "timestamp"]).reset_index(drop=True)

    # Simple weighted fusion: assume upstream frames have prediction columns and average them
    pred_cols = [c for c in merged.columns if c.startswith("prediction")]
    if pred_cols:
        merged["prediction"] = merged[pred_cols].mean(axis=1)

    context.log.info(
        f"Merged data: {len(merged)} rows, columns={list(merged.columns)}",
        source="Custom data processing",
    )
    return merged

from typing import List
import pandas as pd


def main(context, datas: List[pd.DataFrame]) -> pd.DataFrame:
    """Merge several model branches into one wide table.
    datas[0] = BTC model output containing pred_btc
    datas[1] = ETH model output containing pred_eth
    Returns: a wide table with timestamp and each instrument's prediction column
    """
    if not datas:
        context.log.warn("datas is empty; returning an empty DataFrame", source="Custom data processing")
        return pd.DataFrame()

    df_btc = datas[0].copy() if len(datas) > 0 else pd.DataFrame()
    df_eth = datas[1].copy() if len(datas) > 1 else pd.DataFrame()

    # Graceful fallback: return the available branch so the pipeline can continue
    if df_btc.empty:
        return df_eth
    if df_eth.empty:
        return df_btc

    # Keep only the ETH prediction and outer-join it to BTC, retaining every timestamp
    eth_cols = [c for c in ["timestamp", "pred_eth"] if c in df_eth.columns]
    merged = pd.merge(
        df_btc,
        df_eth[eth_cols],
        on="timestamp",
        how="outer",
        suffixes=("", "_eth_dup"),
    )

    # Remove duplicate columns created when both branches contain the same names
    dup_cols = [c for c in merged.columns if c.endswith("_eth_dup")]
    if dup_cols:
        merged.drop(columns=dup_cols, inplace=True)

    merged.sort_values("timestamp", inplace=True)
    merged.reset_index(drop=True, inplace=True)

    context.log.info(
        f"Wide-table merge complete: {len(merged)} rows, columns={list(merged.columns)}",
        source="Custom data processing",
    )
    return merged