SBI証券のCSVをPythonで読み込む｜損益集計まで実装

※本記事には広告・アフィリエイトリンクが含まれます。収益はサイト運営・検証費用に充てています。掲載コードは学習・検証目的です。

先に結論です。証券会社のCSVをpandasで読むとき、詰まるのは文字コードではありません。encoding="cp932"は最初の30秒で解決します。本当に厄介なのは、この4つです。

• ヘッダーの上に説明行が数行入っていて、列名が取れない
• 数量や単価に3桁カンマが入っていて、文字列として読まれる
• 銘柄コードが数値になり、頭の0が消える（1301は無事でも0始まりの銘柄で事故る）
• 買いと売りが同じ列に混在していて、損益を出すには突き合わせが必要

この記事では、読み込みを堅牢にする関数を作り、移動平均法で平均取得単価と実現損益を計算するところまで実装します。最後にyfinanceの現在値と突き合わせて、含み損益まで出します。

📘 外部参考：read_csv（pandas公式） ／ yfinance（GitHub）

CSVのダウンロードと最初の確認

SBI証券のサイトにログイン後、口座管理から取引履歴や保有証券一覧をCSVでダウンロードできます。取得画面やタイミングによって列名も列数も変わるので、決め打ちのコードを書くと次回動きません。まず中身を見ます。

from pathlib import Path

def peek(path: str, n: int = 8, encoding: str = "cp932") -> None:
    with open(path, encoding=encoding, errors="replace") as f:
        for i, line in enumerate(f):
            if i >= n:
                break
            print(f"{i:2d}| {line.rstrip()}")

peek("sbi.csv")

ここで「0行目に日付、1行目が空、2行目から列名」といった構造が見えます。skiprowsに渡す数はここで決めます。

読み込みを堅牢にする

文字コードとヘッダー位置を自動で探し、数値のカンマも処理する関数にまとめます。

import re
import pandas as pd

ENCODINGS = ("cp932", "shift_jis", "utf-8-sig", "utf-8")

def detect_encoding(path: str) -> str:
    for enc in ENCODINGS:
        try:
            with open(path, encoding=enc) as f:
                f.read()
            return enc
        except (UnicodeDecodeError, LookupError):
            continue
    raise ValueError("エンコーディングを判定できません")

def detect_header_row(path: str, encoding: str, keys=("約定日", "銘柄", "取引")) -> int:
    with open(path, encoding=encoding) as f:
        for i, line in enumerate(f):
            if sum(k in line for k in keys) >= 2:
                return i
    return 0

def read_broker_csv(path: str) -> pd.DataFrame:
    enc = detect_encoding(path)
    header = detect_header_row(path, enc)
    df = pd.read_csv(
        path,
        encoding=enc,
        skiprows=header,
        thousands=",",          # 1,234 → 1234
        dtype={"銘柄コード": str},  # 先頭0を守る
        skip_blank_lines=True,
    )
    df.columns = [str(c).strip().replace("\u3000", "") for c in df.columns]
    df = df.dropna(how="all")
    print(f"encoding={enc} header_row={header} rows={len(df)}")
    print("columns:", list(df.columns))
    return df

trades = read_broker_csv("sbi.csv")

dtype={"銘柄コード": str}を必ず入れてください。これがないと、pandasが銘柄コードを整数として読み、後でyfinance用にf"{code}.T"を組み立てるときに桁が足りなくなります。ETFやREITには0から始まるコードもあります。

列名のゆれを吸収する

「約定日」が「約定日時」になっていたり、「取引」が「売買区分」だったりします。候補リストから拾う方式にします。

ALIASES = {
    "date":   ["約定日", "約定日時", "受渡日", "取引日"],
    "code":   ["銘柄コード", "コード", "銘柄・コード"],
    "name":   ["銘柄名", "銘柄"],
    "side":   ["取引", "売買区分", "売買"],
    "qty":    ["数量", "株数", "約定数量"],
    "price":  ["約定単価", "単価", "約定価格"],
    "fee":    ["手数料", "手数料/諸経費等", "手数料等"],
    "amount": ["受渡金額", "受渡金額/決済損益", "約定代金"],
}

def pick_columns(df: pd.DataFrame) -> dict:
    found = {}
    for key, candidates in ALIASES.items():
        for c in candidates:
            if c in df.columns:
                found[key] = c
                break
    missing = [k for k in ("date", "code", "side", "qty", "price") if k not in found]
    if missing:
        raise KeyError(f"必須列が見つかりません: {missing} / 実際の列: {list(df.columns)}")
    return found

def to_number(s: pd.Series) -> pd.Series:
    return pd.to_numeric(
        s.astype(str)
         .str.replace(",", "", regex=False)
         .str.replace("円", "", regex=False)
         .str.replace("−", "-", regex=False)   # 全角マイナス
         .str.strip(),
        errors="coerce",
    )

def normalize_trades(df: pd.DataFrame) -> pd.DataFrame:
    col = pick_columns(df)
    out = pd.DataFrame({
        "date": pd.to_datetime(df[col["date"]], errors="coerce"),
        "code": df[col["code"]].astype(str).str.extract(r"(\d{4})")[0],
        "side": df[col["side"]].astype(str),
        "qty": to_number(df[col["qty"]]),
        "price": to_number(df[col["price"]]),
    })
    out["fee"] = to_number(df[col["fee"]]).fillna(0) if "fee" in col else 0.0
    out["name"] = df[col["name"]] if "name" in col else ""
    out["is_buy"] = out["side"].str.contains("買")
    return out.dropna(subset=["date", "code", "qty", "price"]).sort_values("date").reset_index(drop=True)

tr = normalize_trades(trades)
print(tr.head())

str.extract(r"(\d{4})")で銘柄コードを取り出しているのは、「7203 トヨタ自動車」のように1セルに名前が混ざっている形式に対応するためです。

移動平均法で実現損益を計算する

ここが本題です。日本の特定口座は総平均法に準ずる方法（移動平均法）で取得単価を計算します。買い増すたびに平均単価を更新し、売却時にその平均単価との差額を損益とします。

def realized_pnl(tr: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    positions = {}   # code -> {"qty": 株数, "cost": 平均取得単価}
    records = []

    for _, r in tr.iterrows():
        code, qty, price, fee = r["code"], float(r["qty"]), float(r["price"]), float(r["fee"])
        pos = positions.setdefault(code, {"qty": 0.0, "cost": 0.0})

        if r["is_buy"]:
            total_cost = pos["qty"] * pos["cost"] + qty * price + fee
            pos["qty"] += qty
            pos["cost"] = total_cost / pos["qty"] if pos["qty"] else 0.0
        else:
            if pos["qty"] < qty:
                print(f"警告: {code} の売却数が保有数を超えています（履歴の欠落か信用取引）")
                qty = min(qty, pos["qty"])
            pnl = (price - pos["cost"]) * qty - fee
            records.append({
                "date": r["date"], "code": code, "name": r["name"],
                "qty": qty, "sell_price": price, "avg_cost": round(pos["cost"], 2),
                "pnl": round(pnl, 0),
                "return_pct": round((price / pos["cost"] - 1) * 100, 2) if pos["cost"] else None,
            })
            pos["qty"] -= qty
            if pos["qty"] <= 0:
                pos["qty"], pos["cost"] = 0.0, 0.0

    holdings = pd.DataFrame([
        {"code": c, "qty": p["qty"], "avg_cost": round(p["cost"], 2)}
        for c, p in positions.items() if p["qty"] > 0
    ])
    return pd.DataFrame(records), holdings

closed, holdings = realized_pnl(tr)
print(closed.tail())
print(holdings)

手数料の扱いに注意してください。買いの手数料は取得価額に加算し、売りの手数料は損益から控除します。ここを両方とも単純に引くと、二重計上になります。

集計して振り返る

def summary(closed: pd.DataFrame) -> dict:
    if closed.empty:
        return {}
    wins = closed[closed["pnl"] > 0]
    losses = closed[closed["pnl"] <= 0]
    return {
        "trades": len(closed),
        "total_pnl": int(closed["pnl"].sum()),
        "win_rate": round(len(wins) / len(closed), 3),
        "avg_win": int(wins["pnl"].mean()) if len(wins) else 0,
        "avg_loss": int(losses["pnl"].mean()) if len(losses) else 0,
        "payoff_ratio": round(abs(wins["pnl"].mean() / losses["pnl"].mean()), 2)
                         if len(wins) and len(losses) and losses["pnl"].mean() else None,
        "best": int(closed["pnl"].max()),
        "worst": int(closed["pnl"].min()),
    }

print(summary(closed))

# 月別・銘柄別
print(closed.set_index("date").resample("ME")["pnl"].sum())
print(closed.groupby(["code", "name"])["pnl"].agg(["sum", "count"]).sort_values("sum"))

僕がこれを最初に回したとき、勝率は6割を超えていたのに合計はマイナスでした。avg_winとavg_lossを見たら理由は明らかで、勝ちは小さく利確し、負けは大きく塩漬けにしていました。勝率だけ見ていると気づけないタイプの問題です。

指標	読み方
win_rate	高くても損益がマイナスなら損切りが遅い
payoff_ratio	平均利益÷平均損失。1.0未満なら勝率で稼ぐ型
月別pnl	特定の月に偏っていたら再現性を疑う
銘柄別pnl	1銘柄で全体を説明できるなら実力ではない

現在値と突き合わせて含み損益を出す

import yfinance as yf

def unrealized(holdings: pd.DataFrame) -> pd.DataFrame:
    rows = []
    for _, h in holdings.iterrows():
        t = yf.Ticker(f"{h['code']}.T")
        hist = t.history(period="5d", auto_adjust=False)
        if hist.empty:
            continue
        now = float(hist["Close"].dropna().iloc[-1])
        rows.append({
            "code": h["code"],
            "qty": int(h["qty"]),
            "avg_cost": h["avg_cost"],
            "price": round(now, 1),
            "unrealized": int((now - h["avg_cost"]) * h["qty"]),
            "return_pct": round((now / h["avg_cost"] - 1) * 100, 2),
        })
    df = pd.DataFrame(rows).sort_values("unrealized")
    if not df.empty:
        print("含み損益合計:", int(df["unrealized"].sum()))
    return df

print(unrealized(holdings))

auto_adjust=Falseにしているのは、現在の実際の株価が欲しいからです。調整済み価格は過去の配当を織り込んで下方修正されているので、取得単価との比較には使えません。

複数のCSVをまとめる

import glob

def load_all(pattern: str = "sbi_*.csv") -> pd.DataFrame:
    frames = [normalize_trades(read_broker_csv(f)) for f in sorted(glob.glob(pattern))]
    if not frames:
        raise FileNotFoundError(pattern)
    df = pd.concat(frames, ignore_index=True)
    before = len(df)
    df = df.drop_duplicates(subset=["date", "code", "side", "qty", "price"])
    print(f"重複除去: {before} → {len(df)}")
    return df.sort_values("date").reset_index(drop=True)

期間を重ねてダウンロードすると同じ約定が複数回入ります。drop_duplicatesを入れないと、移動平均の計算が丸ごと狂います。損益がおかしいときは、まずここを疑ってください。

まとめ

証券会社のCSVは、読めればゴールではありません。読めた先に、列名のゆれ、数値のカンマ、銘柄コードの型、重複行、そして買いと売りの突き合わせが待っています。

今回作ったのは4つです。エンコーディングとヘッダー位置を自動判定する読み込み。列名の別名吸収と数値変換。移動平均法での実現損益と平均取得単価。現在値との突き合わせ。

数字が出たら、勝率ではなくpayoff_ratioを先に見てください。自分の取引の癖は、そこに出ます。分析の続きは複数銘柄の比較や最大ドローダウンの計算につなげられます。

---

【投資免責事項】本記事で紹介するコードおよび分析結果は情報提供を目的としたものであり、特定の銘柄への投資を推奨するものではありません。投資判断はご自身の責任で行ってください。