💰
【株】モメンタムの計算など
モメンタム？
アノマリーの一種。↑が詳しそう。開祖は多分↓(1993)。
バックテストしてみた(python)
過去3か月のNASDAQ100のモメンタム上位2銘柄を、3か月おきに乗り換えるバックテスト(20年分)するスクリプト。
CAGR45%とかいうわけわからん値になった。6か月おきだと35%、1か月おきだと15%とかになっちゃう。4半期だといいの？わからん
(AI生成なのでフィジビリもわからん)
# backtest_ndx_momentum_top2.py
# Requirements:
# pip install yfinance pandas numpy requests lxml
import pandas as pd
import numpy as np
import yfinance as yf
import requests
from io import StringIO
import time
# ============= 1) NASDAQ100 ティッカー取得（Wikipedia + UA / StringIO） =============
def get_ndx_tickers_from_wikipedia() -> list:
url = "https://en.wikipedia.org/wiki/Nasdaq-100"
headers = {
"User-Agent": (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/119.0 Safari/537.36"
)
}
try:
res = requests.get(url, headers=headers, timeout=30)
res.raise_for_status()
tables = pd.read_html(StringIO(res.text))
cand = [t for t in tables if any(c.lower() in ["ticker", "symbol"]
for c in t.columns.map(str).str.lower())]
if not cand:
raise RuntimeError("Could not find a table with a 'Ticker' column on the Wikipedia page.")
t = cand[0].copy()
t.columns = [str(c).strip().lower() for c in t.columns]
if "ticker" in t.columns:
tickers = t["ticker"].astype(str).str.strip().tolist()
elif "symbol" in t.columns:
tickers = t["symbol"].astype(str).str.strip().tolist()
else:
tickers = t.iloc[:, 0].astype(str).str.strip().tolist()
# yfinance 表記へ
tickers = [tk.replace(".", "-").upper() for tk in tickers if tk and tk != "—"]
tickers = sorted(set([tk for tk in tickers if tk and tk != "N/A"]))
if not tickers:
raise RuntimeError("Empty ticker list from Wikipedia.")
return tickers
except Exception as e:
print(f"Warning: failed to fetch tickers from Wikipedia ({e}). Using a fallback subset list.")
# 最低限のフォールバック（一部抜粋）
return [
"AAPL","MSFT","AMZN","NVDA","META","GOOG","GOOGL","TSLA","AVGO","PEP",
"COST","ADBE","NFLX","AMD","CSCO","INTC","AMAT","TXN","QCOM","HON",
"SBUX","PDD","AMGN","MDLZ","INTU","BKNG","ADP","PYPL","REGN","MU",
"LRCX","GILD","MELI","ADSK","PANW","ABNB","VRTX","KLAC","FTNT","CRWD"
]
# ============= 2) 1銘柄ずつダウンロード（auto_adjust=True で Close=調整後終値） =============
def download_adjusted_close_per_ticker(tickers, start, end, max_retries=3, pause_sec=0.6) -> pd.DataFrame:
"""
各ティッカーを個別に yfinance.Ticker.history で取得（auto_adjust=True）。
取得できたものだけ列に追加。失敗はスキップ。最後に列方向の完全欠損を落とす。
"""
all_closes = []
failed = []
for idx, tk in enumerate(tickers, 1):
df = None
for attempt in range(1, max_retries + 1):
try:
# auto_adjust=True なので df["Close"] が調整後終値相当
df = yf.Ticker(tk).history(
start=str(start), end=str(end),
auto_adjust=True, actions=False, interval="1d"
)
if df is not None and not df.empty and "Close" in df.columns:
break
except Exception:
pass
time.sleep(pause_sec * attempt) # エクスポネンシャルバックオフ
if df is None or df.empty or "Close" not in df.columns:
failed.append(tk)
continue
s = df["Close"].rename(tk)
all_closes.append(s)
time.sleep(pause_sec) # レート制限回避
if idx % 25 == 0:
print(f" downloaded {idx}/{len(tickers)} tickers...")
if not all_closes:
raise RuntimeError("No ticker data could be downloaded. Check network/proxy/SSL to Yahoo Finance.")
adj = pd.concat(all_closes, axis=1)
# 完全欠損列は削除
adj = adj.dropna(axis=1, how="all")
# タイムゾーンが付いている場合は外す（tz-aware vs tz-naive 比較エラー回避）
if isinstance(adj.index, pd.DatetimeIndex) and getattr(adj.index, "tz", None) is not None:
try:
adj.index = adj.index.tz_convert("UTC").tz_localize(None)
except Exception:
adj.index = adj.index.tz_localize(None)
if failed:
print(f"Warning: failed to download {len(failed)} tickers. Example: {failed[:10]}")
return adj
# ============= 3) バックテスト本体（3MモメンタムTop2、四半期リバランス） =============
def backtest_momentum_top2_quarterly(
tickers,
start_date,
end_date,
lookback_months=3,
hold_months=3,
top_n=2,
):
# モメンタム計算分の余裕を持って前倒しDL開始
dl_start = (pd.Timestamp(start_date) - pd.DateOffset(months=lookback_months + 2)).date()
# 個別取得（調整後終値）
adj = download_adjusted_close_per_ticker(
tickers=tickers,
start=dl_start,
end=end_date,
max_retries=3,
pause_sec=0.6,
)
if adj.empty:
raise RuntimeError("Adjusted Close matrix is empty after per-ticker download.")
# 月末終値（将来非推奨の 'M' を避け 'ME'）
monthly = adj.resample("ME").last().dropna(how="all")
# 指定期間にトリム（比較はタイムゾーン無しで統一）
monthly = monthly.loc[
(monthly.index >= pd.to_datetime(start_date)) &
(monthly.index <= pd.to_datetime(end_date))
]
# 期間が短すぎる場合は情報を出して中断
if len(monthly) < (lookback_months + hold_months + 2):
print("Debug info:")
print(f"- Raw adj earliest date: {adj.index.min() if len(adj.index)>0 else 'NA'}")
print(f"- Raw adj latest date: {adj.index.max() if len(adj.index)>0 else 'NA'}")
print(f"- Monthly rows after trim: {len(monthly)}")
print(f"- Available tickers: {len(monthly.columns)}")
raise RuntimeError("Not enough data after resampling. Verify connectivity or reduce the backtest length.")
period_returns = []
chosen = []
first_now_idx = None
last_next_idx = None
for i in range(lookback_months, len(monthly) - hold_months, hold_months):
now_idx = monthly.index[i]
prev_idx = monthly.index[i - lookback_months]
next_idx = monthly.index[i + hold_months]
px_now = monthly.iloc[i]
px_prev = monthly.iloc[i - lookback_months]
px_next = monthly.iloc[i + hold_months]
# 3か月モメンタム
mom = (px_now / px_prev) - 1.0
mom = mom.dropna()
if mom.empty:
continue
winners = mom.sort_values(ascending=False).head(top_n).index.tolist()
# 等加重の先物期リターン
rets = []
for tk in winners:
p0 = px_now.get(tk, np.nan)
p1 = px_next.get(tk, np.nan)
if np.isfinite(p0) and np.isfinite(p1) and p0 > 0:
rets.append((p1 / p0) - 1.0)
if not rets:
continue
period_ret = float(np.mean(rets))
period_returns.append(period_ret)
chosen.append({"rebalance_date": now_idx, "selected": winners, "period_return": period_ret})
if first_now_idx is None:
first_now_idx = now_idx
last_next_idx = next_idx
if not period_returns:
raise RuntimeError("No valid periods were generated. Check data coverage or tickers.")
equity = np.cumprod(1.0 + np.array(period_returns, dtype=float))
years = (last_next_idx - first_now_idx).days / 365.25
final_value = float(equity[-1])
cagr = final_value ** (1.0 / years) - 1.0
return {
"period_count": len(period_returns),
"cagr": cagr,
"equity_curve": equity,
"period_log": chosen,
"start": first_now_idx,
"end": last_next_idx,
"n_tickers": monthly.shape[1],
}
# ============= 4) エントリポイント =============
if __name__ == "__main__":
# 今日基準で過去20年
end = pd.Timestamp.today().normalize()
start = end - pd.DateOffset(years=20)
print("Fetching NASDAQ-100 tickers from Wikipedia...")
tickers = get_ndx_tickers_from_wikipedia()
print(f"Got {len(tickers)} tickers.")
print("Running backtest (3M momentum, top 2, rebalance quarterly)...")
res = backtest_momentum_top2_quarterly(
tickers=tickers,
start_date=start.date(),
end_date=end.date(),
lookback_months=3,
hold_months=3,
top_n=2
)
print("\n========== Backtest Summary ==========")
print(f"Backtest window: {res['start'].date()} -> {res['end'].date()}")
print(f"Periods (quarters): {res['period_count']}")
print(f"Tickers with usable data: {res['n_tickers']}")
print(f"Average annual return (geometric, CAGR): {res['cagr']:.2%}")
print("\nAll selections: ")
for row in res["period_log"]:
dt = row["rebalance_date"].strftime("%Y-%m-%d")
print(f"{dt}: {row['selected']} -> period return {row['period_return']:.2%}")
直近3か月のモメンタムを出すスクリプト(python)
このスクリプト実行結果の上位2銘柄を3か月おきに乗り換えればよい、ってコト？
(AI生成なのでフィジビリわからん)
# nasdaq100_momentum_recent3month.py
import pandas as pd
import yfinance as yf
import requests
from io import StringIO
from datetime import datetime
import sys
# ====== 設定 ======
WIKI_URL = "https://en.wikipedia.org/wiki/NASDAQ-100"
TOP_N = 20 # 表示件数
TIMEOUT = 20
# 失敗時フォールバック（更新が遅れている可能性はあります）
FALLBACK_TICKERS = [
# アルファベット順（例示。最新構成と異なる場合があります）
"AAPL","MSFT","AMZN","NVDA","GOOGL","GOOG","META","TSLA","AVGO","COST",
"PEP","ADBE","NFLX","AMD","INTC","CSCO","TXN","QCOM","AMAT","INTU",
"BKNG","PDD","LIN","TMUS","MRVL","PYPL","AMGN","SBUX","MU","HON",
"ABNB","REGN","MDLZ","ADI","ISRG","LRCX","PANW","GILD","KLAC","CSX",
"MAR","VRTX","ADI","SNPS","CHTR","CRWD","FTNT","CTAS","NXPI","KDP",
"ADP","ORLY","MELI","CDNS","AZMN" if False else "AZN", # プレースホルダ回避
"AEP","ODFL","ROP","PCAR","KHC","MNST","PDD" if False else "PAYX",
"TEAM","WDAY","XEL","MRNA","EA","EXC","DXCM","ROST","ILMN","LULU",
"CSGP","IDXX","ALGN","BIDU","BIIB","VRSK","CPRT","CTSH","DDOG","ZS",
"SPLK" if False else "FANG", # 旧シンボル対策（使用されない想定）
"DOCU" if False else "FAST","PANW" if False else "PCAR", # ダブり回避
"OKTA" if False else "KDP","BKR","VERX" if False else "VRSN","CEG","ANSS",
"JD","LCID","RIVN","ZM","SIRI","CRWD" if False else "CDW","MDB","ASML","ARM","ON"
][:100] # 100件にトリム
def fetch_nasdaq100_from_wikipedia():
"""WikipediaからNASDAQ-100構成銘柄を取得（User-Agent付き）。"""
headers = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/123.0.0.0 Safari/537.36"
}
resp = requests.get(WIKI_URL, headers=headers, timeout=TIMEOUT)
resp.raise_for_status()
# HTML文字列をread_htmlに食わせる（URL直渡しは403になりやすい）
tables = pd.read_html(StringIO(resp.text))
# 銘柄一覧のテーブルを推定して抽出（列名に Ticker を含むもの）
for tbl in tables:
cols = [str(c).lower() for c in tbl.columns]
if any("ticker" in c for c in cols):
tickers = tbl[[c for c in tbl.columns if "Ticker" in str(c)]].iloc[:,0].dropna().astype(str).str.strip().tolist()
# NASDAQ表記で末尾に注釈が付く場合の掃除
tickers = [t.split(" ")[0].replace(".", "-") for t in tickers]
return tickers
raise RuntimeError("NASDAQ-100 テーブルが見つかりませんでした")
def get_nasdaq100_tickers():
# 1) Wikipediaから取得（推奨）
try:
tks = fetch_nasdaq100_from_wikipedia()
if len(tks) >= 80: # 妥当性チェック
return sorted(set(tks))
except Exception as e:
print(f"[WARN] Wikipediaからの取得に失敗: {e}", file=sys.stderr)
# 2) フォールバック（同梱リスト）
print("[INFO] フォールバックのNASDAQ-100リストを使用します（最新と異なる可能性あり）", file=sys.stderr)
return sorted(set(FALLBACK_TICKERS))
def compute_3m_momentum(tickers):
"""
直近3か月モメンタム = (最新終値 / 3か月前の終値) - 1
- 3か月の定義: カレンダーで厳密に3か月（営業日ギャップは最寄りで代替）
- 価格取得は一括ダウンロードで高速化
"""
# 今日基準。yfinanceは end が非包含なので、+1日相当の余裕を持たせないでもOK
today = pd.Timestamp.today(tz="UTC").tz_localize(None).normalize()
start = (today - pd.DateOffset(months=3)).normalize()
# 4か月分取っておき、3か月前に最も近い営業日を拾う
dl_start = start - pd.DateOffset(days=10)
print(f"[INFO] Downloading OHLCV from {dl_start.date()} to {today.date()} for {len(tickers)} tickers ...", file=sys.stderr)
data = yf.download(
tickers=tickers,
start=dl_start.strftime("%Y-%m-%d"),
end=today.strftime("%Y-%m-%d"),
auto_adjust=True,
progress=False,
interval="1d",
threads=True,
group_by="column",
)
# Close だけ抽出（単一/複数で形が変わるので統一）
close = data["Close"].copy()
if isinstance(close, pd.Series):
close = close.to_frame()
# 3か月前に最も近い営業日の価格を求める
close_sorted = close.sort_index()
if close_sorted.empty:
raise RuntimeError("価格データが空でした。ネットワークやティッカーをご確認ください。")
# 対象日のインデックス（start以降の最初の営業日）
try:
anchor_idx = close_sorted.index.get_indexer([start], method="backfill")[0]
except Exception:
anchor_idx = 0 # 念のため
first_row = close_sorted.iloc[anchor_idx:].iloc[0]
last_row = close_sorted.iloc[-1]
momentum = (last_row / first_row) - 1.0
momentum = momentum.dropna()
out = (
pd.DataFrame({"Ticker": momentum.index, "Momentum": momentum.values})
.sort_values("Momentum", ascending=False)
.reset_index(drop=True)
)
return out
def main():
tickers = get_nasdaq100_tickers()
if not tickers:
raise SystemExit("NASDAQ-100 銘柄の取得に失敗しました。")
# yfinance で価格を取得しモメンタム算出
df = compute_3m_momentum(tickers)
# 表示
pd.set_option("display.float_format", lambda x: f"{x:,.2%}")
print("\n=== NASDAQ-100 モメンタム（直近3か月） 上位 {0} ===".format(TOP_N))
print(df.head(TOP_N).to_string(index=False))
# CSV出力（任意）
out_path = "nasdaq100_momentum_3months.csv"
df.to_csv(out_path, index=False)
print(f"\n[INFO] 全結果を CSV に保存しました: {out_path}")
if __name__ == "__main__":
main()
Discussion