WebAPI連携ミニアプリ：天気×家計簿の自動可視化｜“取る→貯める→見せる”を半日で

weather-budget-app/
  app/
    __init__.py
    config.py        # .env読み込み
    api_client.py    # 天気APIクライアント（丁寧なGET）
    ingest_budget.py # 家計簿CSV取込
    ingest_weather.py# 天気の取得・保存
    repo.py          # SQLiteアクセス（CRUD）
    ui_app.py        # Streamlit UI
  data/
    raw/             # 元CSV（家計簿）
    db.sqlite3       # 永続DB
    cache/           # APIレスポンスキャッシュ
  tests/
  .env.example
  requirements.txt
  README.md

python-dotenv>=1.0
requests>=2.32
pandas>=2.2
sqlalchemy>=2.0
streamlit>=1.36
plotly>=5

API_BASE=https://api.example-weather.com
API_KEY=your_api_key_here
LAT=35.6895
LON=139.6917
TIMEZONE=Asia/Tokyo

from dataclasses import dataclass
from dotenv import load_dotenv
import os

load\_dotenv()

@dataclass(slots=True)
class Settings:
api\_base: str = os.getenv("API\_BASE", "")
api\_key: str | None = os.getenv("API\_KEY")
lat: float = float(os.getenv("LAT", 35.6895))
lon: float = float(os.getenv("LON", 139.6917))
tz: str = os.getenv("TIMEZONE", "Asia/Tokyo")

SET = Settings()

from __future__ import annotations
import json, time, hashlib
from pathlib import Path
import requests
from typing import Any
from .config import SET

CACHE = Path("data/cache"); CACHE.mkdir(parents=True, exist\_ok=True)
UA = "pythonbunseki-weather/1.0 (+[https://pythonbunseki.com](https://pythonbunseki.com))"

class ApiError(Exception): ...

def cache\_path(url: str, params: dict\[str, Any]) -> Path:
key = url + "?" + "&".join(f"{k}={v}" for k, v in sorted(params.items()))
h = hashlib.sha256(key.encode()).hexdigest()\[:16]
return CACHE / f"{h}.json"

def get\_json(path: str, params: dict\[str, Any]) -> dict:
url = SET.api\_base.rstrip("/") + "/" + path.lstrip("/")
params = {\*\*params}
headers = {"User-Agent": UA, "Authorization": f"Bearer {SET.api\_key}"} if SET.api\_key else {"User-Agent": UA}
cp = cache\_path(url, params)
if cp.exists():
return json.loads(cp.read\_text(encoding="utf-8"))
tries = 0
while True:
tries += 1
r = requests.get(url, params=params, headers=headers, timeout=10)
if r.status\_code == 429:
if tries >= 3:
raise ApiError("rate limited")
time.sleep(min(8, 0.5 \* 2 \*\* (tries-1)))
continue
r.raise\_for\_status()
data = r.json()
cp.write\_text(json.dumps(data, ensure\_ascii=False), encoding="utf-8")
return data

import pandas as pd
from pathlib import Path
from sqlalchemy import create_engine

COLS = {"date": "date", "category": "category", "amount": "amount"}

MAP = {
"食費": "食費", "外食": "食費", "カフェ": "食費",
"交通": "交通", "タクシー": "交通",
"日用品": "日用品", "ドラッグストア": "日用品",
}

def read\_budget\_csv(p: Path) -> pd.DataFrame:
df = pd.read\_csv(p, encoding="utf-8", parse\_dates=\[COLS\["date"]])
df = df.rename(columns=COLS)\[\["date", "category", "amount"]]
df\["amount"] = df\["amount"].astype("float")
df\["norm\_cat"] = df\["category"].map(MAP).fillna(df\["category"])
df\["date"] = df\["date"].dt.date
return df

def upsert\_budget(df: pd.DataFrame, db\_path="data/db.sqlite3") -> None:
eng = create\_engine(f"sqlite:///{db\_path}")
df = df.drop\_duplicates(subset=\["date", "norm\_cat", "amount"]).copy()
df.to\_sql("budget", eng, if\_exists="append", index=False)

import pandas as pd
from datetime import date
from sqlalchemy import create_engine
from .config import SET
from .api_client import get_json

# 例: /history?lat=..\&lon=..\&start=YYYY-MM-DD\&end=YYYY-MM-DD

def fetch\_daily\_weather(d0: date, d1: date) -> pd.DataFrame:
js = get\_json("/history", {
"lat": SET.lat, "lon": SET.lon,
"start": d0.isoformat(), "end": d1.isoformat(),
"tz": SET.tz
})
recs = \[]
for r in js.get("days", \[]):
recs.append({
"date": pd.to\_datetime(r\["date"]).date(),
"tavg": float(r.get("tavg", 0)),
"prcp": float(r.get("prcp", 0)),
})
return pd.DataFrame(recs)

def upsert\_weather(df: pd.DataFrame, db\_path="data/db.sqlite3") -> None:
eng = create\_engine(f"sqlite:///{db\_path}")
df.to\_sql("weather", eng, if\_exists="append", index=False)

CREATE TABLE IF NOT EXISTS budget (
  date TEXT, norm_cat TEXT, amount REAL
);
CREATE TABLE IF NOT EXISTS weather (
  date TEXT PRIMARY KEY, tavg REAL, prcp REAL
);

import pandas as pd
from sqlalchemy import create_engine

ENG\_STR = "sqlite:///data/db.sqlite3"

def load\_joined() -> pd.DataFrame:
eng = create\_engine(ENG\_STR)
q = """
SELECT b.date, b.norm\_cat AS category, b.amount,
w\.tavg, w\.prcp
FROM budget b
LEFT JOIN weather w USING(date)
"""
return pd.read\_sql(q, eng, parse\_dates=\["date"])

def agg\_monthly(df: pd.DataFrame) -> pd.DataFrame:
g = df.copy()
g\["ym"] = g\["date"].dt.to\_period("M").astype(str)
return g.groupby(\["ym", "category"], as\_index=False).agg(
spend=("amount","sum"),
prcp=("prcp","mean"),
tavg=("tavg","mean")
)

import streamlit as st
import pandas as pd
import plotly.express as px
from app.repo import load_joined, agg_monthly

st.set\_page\_config(page\_title="天気×家計簿", layout="wide")
st.title("天気×家計簿の自動可視化")

@st.cache\_data(ttl=300)
def get\_data():
df = load\_joined()
return df.dropna(subset=\["amount"])

df = get\_data()
if df.empty:
st.warning("データがありません。家計簿CSV取り込みと天気取得を先に実行してください。")
st.stop()

cats = sorted(df\["category"].dropna().unique().tolist())
sel = st.multiselect("カテゴリを選択", cats, default=cats\[:3])
period = st.slider("期間（月）", 3, 24, 12)
latest = df\["date"].max()
from\_date = (latest - pd.DateOffset(months=period)).normalize()

f = df\[(df\["category"].isin(sel)) & (df\["date"] >= from\_date)]
monthly = agg\_monthly(f)

col1, col2 = st.columns(2)
with col1:
st.subheader("月次支出（カテゴリ別）")
fig1 = px.bar(monthly, x="ym", y="spend", color="category", barmode="stack")
fig1.update\_layout(showlegend=True, xaxis\_title="月", yaxis\_title="支出")
st.plotly\_chart(fig1, use\_container\_width=True)
with col2:
st.subheader("平均気温と降水量（同月平均）")
fig2 = px.line(monthly.groupby("ym", as\_index=False).mean(numeric\_only=True), x="ym", y=\["tavg","prcp"])
fig2.update\_layout(xaxis\_title="月", yaxis\_title="値")
st.plotly\_chart(fig2, use\_container\_width=True)

st.subheader("支出×天気の関係（散布図）")
sc = monthly.groupby("ym", as\_index=False).agg(spend=("spend","sum"), prcp=("prcp","mean"), tavg=("tavg","mean"))
fig3 = px.scatter(sc, x="tavg", y="spend", size="prcp", hover\_name="ym")
fig3.update\_layout(xaxis\_title="平均気温", yaxis\_title="総支出", legend\_title="降水量(点サイズ)")
st.plotly\_chart(fig3, use\_container\_width=True)

st.caption("注：気象データは地点平均。月内のイベント（ボーナス/旅行等）は別途注釈を。")

# 天気を毎朝6:00に更新
0 6 * * * cd /path/to/weather-budget-app && /usr/bin/python -m app.ingest_weather >> logs/$(date +\%Y\%m\%d).log 2>&1