-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_sp500_only.py
More file actions
54 lines (42 loc) · 1.98 KB
/
Copy pathtest_sp500_only.py
File metadata and controls
54 lines (42 loc) · 1.98 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import pandas as pd
import requests
from io import StringIO
def test_sp500_backtrack():
print("Testing dynamic S&P 500 Wikipedia parsing with MultiIndex flattening...")
url = "https://en.wikipedia.org/wiki/List_of_S%26P_500_companies"
headers = {"User-Agent": "Mozilla/5.0"}
response = requests.get(url, headers=headers)
if response.status_code != 200:
print(f"Failed to fetch: HTTP {response.status_code}")
return
tables = pd.read_html(StringIO(response.text))
df_current = tables[0]
df_changes = tables[1]
# Flatten MultiIndex columns: e.g. ('Added', 'Ticker') -> 'Added_Ticker'
df_changes.columns = [f"{c[0]}_{c[1]}" if isinstance(c, tuple) else c for c in df_changes.columns]
current_symbols = set(df_current['Symbol'].dropna().tolist())
# Use the flattened date column 'Effective Date_Effective Date'
date_col = 'Effective Date_Effective Date'
df_changes[date_col] = pd.to_datetime(df_changes[date_col], errors='coerce')
df_changes = df_changes.dropna(subset=[date_col]).sort_values(by=date_col, ascending=False)
cutoff = pd.to_datetime("2020-01-01")
symbols = set(current_symbols)
for idx, row in df_changes.iterrows():
if row[date_col] < cutoff:
break
# added and removed columns are uniquely named now
added = str(row.get('Added_Ticker') or '').strip()
removed = str(row.get('Removed_Ticker') or '').strip()
if added and added != 'nan' and added != '':
symbols.discard(added)
if removed and removed != 'nan' and removed != '':
symbols.add(removed)
print(f"✅ Success! Reconstructed S&P 500 symbols as of 2020-01-01: {len(symbols)}")
print(f"Sample of 10 symbols: {list(symbols)[:10]}")
if __name__ == "__main__":
try:
test_sp500_backtrack()
except Exception as e:
print(f"❌ Failed: {e}")
import traceback
traceback.print_exc()