Hi Dev.Teaam ,
May I request for your assistance on this Client's query please? Thank you.
eikon_data_extractor.py (v4)
import datetime as dt
import pandas as pd
import eikon as ek
---------------------------------------------------------------------1. Conexión (usa tu App Key)
ek.set_app_key("")
ek.set_timeout(120)
Si tu sesión de Workspace ya está autenticada, puedes comentar la línea anterior.---------------------------------------------------------------------2. Parámetros
ASSET_RIC = ".SPX"
MATURITY = "2025-07-18"
START_HIST = "2015-01-01"
END_HIST = dt.date.today()
BATCH = 250
---------------------------------------------------------------------3. Utilidad para guardar CSV
def to_csv(df: pd.DataFrame, filename: str) -> None:
df.to_csv(filename, index=False)
print(f"✅ Guardado {filename}")
---------------------------------------------------------------------4. Localizar un chain-RIC que no devuelva error
def find_working_chain(asset_ric: str) -> str:
candidates = [
f"0#{asset_ric}", # 0#.SPX
f"{asset_ric}", # .SPX
f"{asset_ric}w*", # .SPXw*
f"0#{asset_ric}w*", # 0#.SPXw*
f"{asset_ric}.U", # .SPX.U
f"0#{asset_ric}.U", # 0#.SPX.U
]
for ric in candidates:
test_df, err = ek.get_data(ric, ["TR.RIC"])
if err is None and not test_df.empty:
print(f" Usando cadena: {ric}")
return ric
raise RuntimeError(
"Ninguno de los patrones de cadena RIC ha sido reconocido.
"
"Comprueba tu licencia o prueba con otro activo."
)
---------------------------------------------------------------------5. Cadena de opciones + griegas
def fetch_option_chain(asset_ric: str, maturity: str) -> pd.DataFrame:
chain_ric = find_working_chain(asset_ric)
fields = [
"TR.RIC", "TR.OptionExpirationDate", "TR.OptionType",
"TR.OptionStrike", "TR.ImpliedVolatility",
"TR.GreekDelta", "TR.GreekGamma", "TR.GreekVega", "TR.GreekTheta",
"TR.Bid", "TR.Ask", "TR.Volume"
]
df, err = ek.get_data(chain_ric, fields)
if err:
raise RuntimeError(err)
maturity_dt = pd.to_datetime(maturity)
df = df[pd.to_datetime(df["Option Expiration Date"]) == maturity_dt]
if df.empty:
raise RuntimeError(
f"No se encontraron opciones con vencimiento {maturity} "
f"en la cadena {chain_ric}."
)
return df.reset_index(drop=True)
---------------------------------------------------------------------6. Spot del subyacente
def fetch_spot(asset_ric: str) -> pd.DataFrame:
spot_df, err = ek.get_data(asset_ric, ["TRDPRC_1", "BID", "ASK"])
if err:
raise RuntimeError(err)
return spot_df
---------------------------------------------------------------------7. Curva libre de riesgo
def fetch_risk_free(start: str, end: str) -> pd.DataFrame:
rf_df = ek.get_timeseries("USGG3M=RR", start_date=start,
end_date=end, interval="daily")
rf_df = (rf_df.rename(columns={"CLOSE": "rf_rate"})
.reset_index()
.rename(columns={"Date": "date"}))
return rf_df
---------------------------------------------------------------------8. Series históricas
def fetch_histories(asset_ric: str, start: str, end: str):
asset_hist = ek.get_timeseries(asset_ric, start_date=start,
end_date=end, fields="CLOSE",
interval="daily")
vix_hist = ek.get_timeseries(".VIX", start_date=start,
end_date=end, fields="CLOSE",
interval="daily")
asset_hist = (asset_hist.rename(columns={"CLOSE": "close"})
.reset_index()
.rename(columns={"Date": "date"}))
vix_hist = (vix_hist.rename(columns={"CLOSE": "vix"})
.reset_index()
.rename(columns={"Date": "date"}))
return asset_hist, vix_hist
---------------------------------------------------------------------9. Ejecución principal
if name == "main":
9-a Opciones
opt_df = fetch_option_chain(ASSET_RIC, MATURITY)
to_csv(opt_df, f"options_{ASSET_RIC.strip('.')}_{MATURITY}.csv")
9-b Spot
spot_df = fetch_spot(ASSET_RIC)
to_csv(spot_df, f"spot_{ASSET_RIC.strip('.')}.csv")
9-c Curva libre de riesgo
rf_df = fetch_risk_free(START_HIST, END_HIST)
to_csv(rf_df, "risk_free_curve.csv")
9-d Series históricas
asset_hist, vix_hist = fetch_histories(ASSET_RIC, START_HIST, END_HIST)
to_csv(asset_hist, f"history_{ASSET_RIC.strip('.')}.csv")
to_csv(vix_hist, "history_VIX.csv")
Client is using it wirhout API Key
