Streamlit で株価ビューワーを作る

1
"""app.py — J-Quants 連携の株価ビューワー
2
"""
3
from __future__ import annotations
4

5
import os
6
from datetime import date, timedelta
7

8
import pandas as pd
9
import requests
10
import streamlit as st
11
from dotenv import load_dotenv
12
from plotly.subplots import make_subplots
13
import plotly.graph_objects as go
14

15
load_dotenv()
16
API_KEY = os.environ["JQUANTS_API_KEY"]
17
JQ_BASE = "https://api.jquants.com/v2"
18

19

20
# --- J-Quants クライアント -----------------------------------------------------
21

22
@st.cache_data(ttl=60 * 30)
23
def fetch_prices(code: str, start: date, end: date) -> pd.DataFrame:
24
    """銘柄 1 つの日次株価を J-Quants から取得する。"""
25
    headers = {"x-api-key": API_KEY}
26
    params = {"code": code, "from": start.isoformat(), "to": end.isoformat()}
27
    rows: list[dict] = []
28
    pagination_key: str | None = None
29
    while True:
30
        if pagination_key:
31
            params["pagination_key"] = pagination_key
32
        res = requests.get(
33
            f"{JQ_BASE}/equities/bars/daily", headers=headers, params=params, timeout=15,
34
        )
35
        res.raise_for_status()
36
        data = res.json()
37
        rows.extend(data["data"])
38
        pagination_key = data.get("pagination_key")
39
        if not pagination_key:
40
            break
41

42
    df = pd.DataFrame(rows)
43
    if df.empty:
44
        return df
45
    df["Date"] = pd.to_datetime(df["Date"])
46
    df = df.sort_values("Date").reset_index(drop=True)
47
    return df
48

49

50
# --- チャート --------------------------------------------------------------
51

52
def build_chart(df: pd.DataFrame, sma_short: int, sma_long: int) -> go.Figure:
53
    df = df.copy()
54
    df[f"sma{sma_short}"] = df["C"].rolling(sma_short).mean()
55
    df[f"sma{sma_long}"]  = df["C"].rolling(sma_long).mean()
56

57
    fig = make_subplots(
58
        rows=2, cols=1,
59
        shared_xaxes=True,
60
        row_heights=[0.75, 0.25],
61
        vertical_spacing=0.03,
62
    )
63
    fig.add_trace(go.Candlestick(
64
        x=df["Date"], open=df["O"], high=df["H"],
65
        low=df["L"], close=df["C"], name="OHLC", showlegend=False,
66
    ), row=1, col=1)
67
    fig.add_trace(go.Scatter(x=df["Date"], y=df[f"sma{sma_short}"],
68
                             name=f"SMA{sma_short}", line=dict(width=1)), row=1, col=1)
69
    fig.add_trace(go.Scatter(x=df["Date"], y=df[f"sma{sma_long}"],
70
                             name=f"SMA{sma_long}", line=dict(width=1)), row=1, col=1)
71
    fig.add_trace(go.Bar(x=df["Date"], y=df["Vo"], name="Vo",
72
                         marker=dict(line=dict(width=0)), showlegend=False),
73
                  row=2, col=1)
74
    fig.update_layout(
75
        height=600,
76
        xaxis_rangeslider_visible=False,
77
        margin=dict(l=40, r=20, t=20, b=20),
78
        template="plotly_white",
79
    )
80
    return fig
81

82

83
# --- アプリ本体 -------------------------------------------------------------
84

85
st.set_page_config(page_title="株価ビューワー", layout="wide")
86
st.title("株価ビューワー")
87

88
with st.sidebar:
89
    st.header("設定")
90
    code = st.text_input("銘柄コード", value="7203")
91
    days = st.slider("表示期間(営業日換算)", 60, 1000, 300, step=20)
92
    sma_short = st.slider("短期 SMA", 5, 60, 25, step=5)
93
    sma_long = st.slider("長期 SMA", 20, 200, 75, step=5)
94

95
end = date.today()
96
start = end - timedelta(days=int(days * 1.5))   # 営業日換算で広めに取る
97

98
with st.spinner("J-Quants から取得しています..."):
99
    df = fetch_prices(code, start, end)
100

101
if df.empty:
102
    st.warning("データが取得できませんでした。コードと期間を確認してください。")
103
    st.stop()
104

105
# 表示用に末尾だけ
106
df_view = df.tail(days).reset_index(drop=True)
107

108
latest = df_view.iloc[-1]
109
prev = df_view.iloc[-2]
110
pct = (latest["C"] / prev["C"] - 1) * 100
111

112
col1, col2, col3 = st.columns(3)
113
col1.metric("最新終値", f"{latest['C']:.0f} 円", f"{pct:+.2f}%")
114
col2.metric("最新出来高", f"{latest['Vo']:,.0f}")
115
col3.metric("最新日付", latest["Date"].strftime("%Y-%m-%d"))
116

117
st.plotly_chart(build_chart(df_view, sma_short, sma_long), use_container_width=True)
118

119
st.subheader("直近 30 日")
120
st.dataframe(df_view.tail(30)[["Date", "O", "H", "L", "C", "Vo"]])

1
try:
2
    df = fetch_prices(code, start, end)
3
except requests.HTTPError as e:
4
    st.error(f"取得に失敗しました: {e.response.status_code}")
5
    st.stop()
6
except requests.Timeout:
7
    st.error("J-Quants の応答がタイムアウトしました")
8
    st.stop()

目的:
J-Quants から日次株価を取得し、Streamlit でローソク足ビューワーを表示する。

入力:
- 銘柄コード(text_input)
- 表示期間(slider, 60-1000 営業日)
- 短期/長期 SMA 期間(slider)

要件:
- J-Quants の id_token は refresh_token から取得し @st.cache_data でキャッシュ
- 株価取得関数も @st.cache_data でキャッシュ(ttl=30 分)
- ローソク足 + 出来高(2 段)、Plotly の make_subplots
- 上部に最新終値・前日比・最新日付のメトリクス
- 直近 30 日のテーブル
- エラー時は st.error → st.stop

制約:
- streamlit 1.40 系 / pandas 2.2 系 / plotly 5.20 系
- 認証情報は .env から python-dotenv で読み込み
- ファイル名は app.py