import refinitiv.data as rd
import pandas as pd
── Config ───────────────────────────────────────────────────────────────────
START_DATE = "2023-01-01"
END_DATE = "2026-06-10"
MONTH_CODES = {
'F': 1, 'G': 2, 'H': 3, 'J': 4,
'K': 5, 'M': 6, 'N': 7, 'Q': 8,
'U': 9, 'V': 10, 'X': 11, 'Z': 12,
}
MONTH_NAMES = {
1: 'Jan', 2: 'Feb', 3: 'Mar', 4: 'Apr',
5: 'May', 6: 'Jun', 7: 'Jul', 8: 'Aug',
9: 'Sep', 10: 'Oct', 11: 'Nov', 12: 'Dec',
}
Strictly restricted to F, G, H, J, K only for Year 26
EXPIRED_26_CODES = {'F', 'G', 'H', 'J', 'K'}
── Build Native Spread RICs ─────────────────────────────────────────────────
def build_spread_market_list():
"""
Generates native spread RICs directly matching the user's updated rules:
- If Year is 26 and Near month is strictly F through K, append '^2'
- All other months (including M and N) and years pass without suffix.
"""
# 1. Create an ordered chronological list of all outright legs
legs = []
for year in range(26, 29): # 26, 27, 28
for code, month in sorted(MONTH_CODES.items(), key=lambda x: x[1]):
legs.append((year, code))
spread_map = {}<br>
# 2. Build adjacent spreads with a gap of 1, 2, and 3 months
for i, (y1, c1) in enumerate(legs):
for gap in range(1, 4):
if i + gap >= len(legs):
break
y2, c2 = legs[i + gap]<br>
# Construct base spread code format: FCPOF26-G26
spread_ric = f"FCPO{c1}{y1}-{c2}{y2}"<br>
# UPDATED RULE: Apply ^2 suffix ONLY if Year 26 and month is F, G, H, J, or K
if y1 == 26 and c1 in EXPIRED_26_CODES:
spread_ric += "^2"<br>
# Create standard display label: e.g., "Jan26-Feb26"
m1_name = MONTH_NAMES[MONTH_CODES[c1]]
m2_name = MONTH_NAMES[MONTH_CODES[c2]]
clean_label = f"{m1_name}{y1}-{m2_name}{y2}"<br>
spread_map[spread_ric] = clean_label<br>
return spread_map<br>
── Main ─────────────────────────────────────────────────────────────────────
print("Opening Refinitiv session...")
rd.open_session()
Generate our target spreads and mapping dictionary
spread_lookup = build_spread_market_list()
all_spread_rics = list(spread_lookup.keys())
print(f"Total Native Spreads to fetch ({len(all_spread_rics)})")
print(f"Sample Tickers: {all_spread_rics[:5]} ... {all_spread_rics[-5:]}\n")
Fetch one by one
all_dfs = []
for i, ric in enumerate(all_spread_rics):
print(f"Fetching RIC {i + 1}/{len(all_spread_rics)}: {ric}")
try:
df_ric = rd.get_data(
universe=ric,
fields=[
f"TR.CLOSEPRICE(SDate={START_DATE},EDate={END_DATE}).date",
f"TR.CLOSEPRICE(SDate={START_DATE},EDate={END_DATE})"
]
)
if df_ric is not None and not df_ric.empty:
all_dfs.append(df_ric)
else:
print(f"Warning: RIC {ric} returned empty data.")<br>
except Exception as e:
print(f"Warning: RIC {ric} failed - {e}")
continue<br>
print("\nProcessing and cleaning direct spread data...")
if all_dfs:
raw = pd.concat(all_dfs, ignore_index=True)
# Standardize columns
raw.columns = ["Instrument", "Date", "Close"]
raw = raw.dropna(subset=["Date", "Close"])
# Format fields
raw["Date"] = pd.to_datetime(raw["Date"]).dt.date
raw["Value"] = pd.to_numeric(raw["Close"], errors="coerce")
raw = raw.dropna(subset=["Value"])<br>
# Map raw RIC names to your requested clean output labels ("Jan26-Feb26")
raw["Contract"] = raw["Instrument"].map(spread_lookup)
# Finalize structure
out_df = raw[["Date", "Contract", "Value"]].copy()
out_df["Value"] = out_df["Value"].round(2)
# Sort chronologically
out_df = out_df.sort_values(["Date", "Contract"]).reset_index(drop=True)<br>
print(f"\nTotal spread rows fetched: {len(out_df)}")
print(f"Unique spreads processed: {out_df['Contract'].nunique()}")
print("\nFirst 30 rows of output:")
print(out_df.head(30))<br>
# Save output
out_path = "fcpo_native_spreads1.csv"
out_df.to_csv(out_path, index=False)
print(f"\nSaved directly to: {out_path}")<br>
else:
print("\nNo data was successfully fetched from Refinitiv.")
rd.close_session()
I gave the client the below sample code
import refinitiv.data as rd
rd.open_session()
df = rd.get_history(
universe = ['FCPOc1'],
fields = ['TR.CLOSEPRICE(Frq=D,SDate=2023-01-01,EDate=2026-06-11)']
)
display(df)
df.to_excel("filename.xlsx")
Since he wanted to export the data to .xlsx
Need your inputs on what might be the reason from the Code's prespective.
Will check with content if the code is correct from our end.
