import pandas as pd import talib import numpy as np def talib_df(df): """ Return a single DataFrame containing: - Original Date + OHLCV columns - All numeric TA-Lib indicators - All CDL patterns (0/1) """ df = df.copy() print(df) # Ensure OHLCV columns exist for col in ['Open','High','Low','Close','Volume']: if col not in df.columns: raise ValueError(f"Missing column: {col}") df=df.reset_index() # Base DF with Date + OHLCV result_df = df[['Date','Close','High','Low','Open','Volume']].copy() # --- Indicators --- df_std = df.copy() df_std.columns = [c.lower() for c in df_std.columns] ohlcv = {k: df_std.get(k) for k in ['open','high','low','close','volume']} indicator_list = [ f for f in dir(talib) if not f.startswith("CDL") and not f.startswith("_") and f not in ["wraps", "wrapped_func"] ] indicator_dfs = [] for name in indicator_list: func = getattr(talib, name) try: if ohlcv['close'] is None: continue res = func(ohlcv['close'].values.astype(float)) if isinstance(res, tuple): for i, arr in enumerate(res): indicator_dfs.append(pd.DataFrame(arr, index=df.index, columns=[f"{name}_{i}"])) else: indicator_dfs.append(pd.DataFrame(res, index=df.index, columns=[name])) except: continue if indicator_dfs: result_df = pd.concat([result_df] + indicator_dfs, axis=1) # --- CDL Patterns --- pattern_list = [f for f in dir(talib) if f.startswith("CDL")] for p in pattern_list: func = getattr(talib, p) try: res = func( df['Open'].values.astype(float), df['High'].values.astype(float), df['Low'].values.astype(float), df['Close'].values.astype(float) ) result_df[p] = (res != 0).astype(int) except: continue return result_df