リレーショナルデータベースとPythonの連携

基本概念

リレーショナルデータベース（以下、RDB）は、データを表形式で管理し、表同士の関係性を定義することで、効率的なデータ操作を可能にするデータベースシステムである。RDBでは、SQL（Structured Query Language）を用いてデータの検索、挿入、更新、削除を行う。本文書では、RDBの基本構造を理解した上で、Pythonと連携した応用例を学ぶ。

データベースの構造と設計

RDBは、行（レコード）と列（フィールド）から構成されるテーブルを基本単位とする。各テーブルは、列ごとにデータ型と制約を定義し、データの整合性を保つ。

テーブル定義の例

以下にテーブル定義SQLの実装例を示す。

CREATE TABLE product (
    id INTEGER PRIMARY KEY,
    product_name TEXT NOT NULL,
    type TEXT,
    cost REAL,
    created_at DATETIME
);

データ型

SQLiteで使用する主要なデータ型は以下の通りである。

• INTEGER（整数値を格納）
• REAL（浮動小数点数を格納）
• TEXT（文字列を格納）
• DATETIME（日時情報を格納）

制約

データの整合性を保つための制約は以下の通りである。

• NOT NULL（null値を許容しない必須項目）
• UNIQUE（重複を許容しない一意性制約）
• PRIMARY KEY（テーブルの主キー（重複のない一意の識別子））

開発環境の構築

Windows上でPythonと必要なライブラリをインストールする手順を説明する。

1. Python 3.12のインストール

インストール済みの場合は実行不要である。

管理者権限でコマンドプロンプトを起動（手順：Windowsキーまたはスタートメニュー > cmd と入力 > 右クリック > 「管理者として実行」）し、以下を実行する。管理者権限は、wingetの--scope machineオプションでシステム全体にソフトウェアをインストールするために必要である。

REM Python をシステム領域にインストール
winget install --scope machine --id Python.Python.3.12 -e --silent
REM Python のパス設定
set "PYTHON_PATH=C:\Program Files\Python312"
set "PYTHON_SCRIPTS_PATH=C:\Program Files\Python312\Scripts"
if exist "%PYTHON_PATH%" setx PYTHON_PATH "%PYTHON_PATH%" /M >nul
if exist "%PYTHON_SCRIPTS_PATH%" setx PYTHON_SCRIPTS_PATH "%PYTHON_SCRIPTS_PATH%" /M >nul
echo "%PATH%" | find /i "%PYTHON_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_PATH%" /M >nul
echo "%PATH%" | find /i "%PYTHON_SCRIPTS_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_SCRIPTS_PATH%" /M >nul

【関連する外部ページ】

Pythonの公式ページ: https://www.python.org/

2. 必要なPythonライブラリのインストール

1. Windowsで、管理者権限でコマンドプロンプトを起動（手順：Windowsキーまたはスタートメニュー > cmd と入力 > 右クリック > 「管理者として実行」）する。
2. 以下のコマンドを実行し、必要なライブラリをインストールする。

   pip install -U yfinance pandas numpy scikit-learn statsmodels geopandas matplotlib japanize-matplotlib requests

【関連する外部ページ】

• Python公式サイト：https://www.python.org/

【サイト内の関連ページ】

• Python詳細ガイド：別ページ »

データベース操作の基本

以下のPythonプログラムは、データベース操作を管理するクラスの実装例である。このクラスは、データベースの初期化、データの挿入、クエリの実行を処理する。クラス化により、接続管理やエラー処理を一箇所にまとめ、コードの再利用性を高めている。

import sqlite3
import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta

class DatabaseManager:
    def __init__(self, db_path):
        """データベース接続を初期化する"""
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
        self.initialize_database()

    def initialize_database(self):
        """テーブルを作成し、サンプルデータを取得して格納する"""
        # 株価テーブルの作成
        self.cursor.execute("""
        CREATE TABLE IF NOT EXISTS stock_prices (
            date TEXT,
            ticker TEXT,
            open REAL,
            high REAL,
            low REAL,
            close REAL,
            volume INTEGER,
            PRIMARY KEY (date, ticker)
        )
        """)

        # 日経平均とTOPIX連動ETFのデータを取得
        tickers = ['^N225', '1306.T']
        end_date = datetime.now()
        start_date = end_date - timedelta(days=30)

        for ticker in tickers:
            # multi_level_index=Falseで単一レベルの列名を取得
            data = yf.download(ticker, start=start_date, end=end_date, multi_level_index=False)
            data['ticker'] = ticker
            data.reset_index(inplace=True)

            # データの挿入
            for _, row in data.iterrows():
                self.insert_data('stock_prices', [
                    row['Date'].strftime('%Y-%m-%d'),
                    row['ticker'],
                    row['Open'],
                    row['High'],
                    row['Low'],
                    row['Close'],
                    row['Volume']
                ])

    def execute_query(self, query, params=None):
        """SQLクエリを実行し、結果をDataFrame（pandasライブラリの表形式データ構造）として返す"""
        if params:
            return pd.read_sql_query(query, self.conn, params=params)
        return pd.read_sql_query(query, self.conn)

    def insert_data(self, table, data):
        """指定したテーブルにデータを挿入する"""
        try:
            placeholders = ','.join(['?' for _ in data])
            query = f"INSERT OR REPLACE INTO {table} VALUES ({placeholders})"
            self.cursor.execute(query, data)
            self.conn.commit()
        except sqlite3.Error as e:
            print(f"データ挿入エラー: {e}")

    def get_latest_prices(self):
        """最新の価格データを取得する"""
        query = """
        SELECT date, ticker, close
        FROM stock_prices
        WHERE date = (SELECT MAX(date) FROM stock_prices)
        """
        return self.execute_query(query)

    def close(self):
        """データベース接続を閉じる"""
        self.conn.close()

実行手順

上記のクラス定義と以下の実行コードを同じPythonファイル（例：db_manager.py）に保存し、実行する。

# データベースの作成と初期化
db = DatabaseManager('stocks.db')

# 最新価格の取得
latest = db.get_latest_prices()
print(latest)

# 接続を閉じる
db.close()

実行結果の確認

正常に実行されると、以下のような出力が表示される。日付と価格は実行時点のデータにより異なる。

実行後、プログラムと同じディレクトリに stocks.db ファイルが作成される。このファイルがSQLiteデータベースである。

応用例

RDBとPythonを組み合わせた3つの応用例を紹介する。為替データの管理、時系列データの分析、地理空間データの活用を順に解説する。

1. 為替データの管理と分析

為替データを管理するためのテーブル定義を以下に示す。このテーブルは、複数の通貨の為替レートを時系列で記録する構造を持つ。

CREATE TABLE quote (
    seq INTEGER PRIMARY KEY NOT NULL,
    at DATETIME,
    USD REAL,
    JPY REAL,
    EUR REAL
);

以下のPythonプログラムは、Yahoo Finance APIを使用して為替データを取得し、データベースに保存する実装例である。

import pandas as pd
import sqlite3
from datetime import datetime, timedelta
import yfinance as yf

def analyze_exchange_rates():
    # 為替データのダウンロード（Yahoo Finance APIを使用）
    currency_pairs = ['USDJPY=X', 'EURJPY=X']
    end_date = datetime.now()
    start_date = end_date - timedelta(days=30)

    # データ取得と整形
    dfs = []
    for pair in currency_pairs:
        # multi_level_index=Falseで単一レベルの列名を取得
        df = yf.download(pair, start=start_date, end=end_date, multi_level_index=False)
        # 通貨ペア名から通貨コードを抽出（USDJPY=X -> USD, EURJPY=X -> EUR）
        currency_code = pair.replace('JPY=X', '')
        df = df['Close'].rename(currency_code)
        dfs.append(df)

    # データフレームの結合
    rates_df = pd.concat(dfs, axis=1)
    rates_df['JPY'] = 1  # 基準通貨

    # データベースへの保存
    conn = sqlite3.connect('forex.db')
    rates_df.to_sql('quote', conn, if_exists='replace')

    # 分析実行
    stats = rates_df.describe()
    changes = rates_df.pct_change().mean()

    conn.close()
    return stats, changes

このプログラムは、Yahoo Finance APIから為替データを取得し、pandasライブラリによる統計分析を実行する。30日間の時系列データを取得し、基本統計量と日次変動率を算出する。SQLiteデータベースにデータを保存することで、継続的な分析が可能な形式で保存する。

【注意】Yahoo Finance APIはインターネット接続が必要であり、取得できるデータや頻度に制限がある場合がある。

実行手順

上記の関数定義と以下の実行コードを同じPythonファイル（例：forex_analysis.py）に保存し、実行する。

stats, changes = analyze_exchange_rates()
print("基本統計量:")
print(stats)
print("\n日次変動率:")
print(changes)

実行結果の確認

正常に実行されると、以下のような出力が表示される。数値は実行時点のデータにより異なる。

実行後、プログラムと同じディレクトリに forex.db ファイルが作成される。

2. 時系列データの分析と予測

以下のPythonプログラムは、株価データを使用した時系列分析と異常値検出を行う実装例である。ARIMA（自己回帰和分移動平均）モデルによる予測とIsolation Forest（データの分布から外れた異常値を検出する機械学習手法）による異常値検出を組み合わせている。

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from statsmodels.tsa.arima.model import ARIMA
import yfinance as yf
import matplotlib.pyplot as plt
import japanize_matplotlib

def analyze_timeseries(ticker='7203.T', period='1y', forecast_periods=10):
    """
    株価データを使用した時系列分析
    ticker: 株式コード（デフォルトはトヨタ自動車）
    period: 取得期間
    forecast_periods: 予測期間
    """
    # Yahoo Financeからデータ取得
    stock = yf.Ticker(ticker)
    data = stock.history(period=period)

    # 終値を使用
    close_prices = data['Close'].values

    # ARIMAモデルによる予測
    model = ARIMA(close_prices, order=(1, 1, 1))
    model_fit = model.fit()
    forecast = model_fit.forecast(steps=forecast_periods)

    # 異常値検出
    iso_forest = IsolationForest(contamination=0.1, random_state=42)
    anomalies = iso_forest.fit_predict(close_prices.reshape(-1, 1))

    # 結果の可視化
    plt.figure(figsize=(12, 6))
    plt.plot(close_prices, label='実績値')
    plt.plot(range(len(close_prices), len(close_prices) + forecast_periods),
             forecast, label='予測値', linestyle='--')
    plt.scatter(np.where(anomalies == -1)[0],
                close_prices[anomalies == -1],
                color='red', label='異常値')
    plt.title(f'{ticker}の株価分析')
    plt.legend()

    results = {
        'forecast': forecast,
        'anomalies': close_prices[anomalies == -1],
        'anomaly_dates': data.index[anomalies == -1]
    }

    return results

このプログラムは、Yahoo Financeから取得した株価データに対し、ARIMAモデルによる時系列予測とIsolation Forestによる異常値検出を実装する。予測結果と異常値を可視化し、金融データ分析の手法を組み合わせている。

実行手順

上記の関数定義と以下の実行コードを同じPythonファイル（例：timeseries_analysis.py）に保存し、実行する。

results = analyze_timeseries('7203.T', '1y', 10)
print("予測値:", results['forecast'])
print("異常値検出日:", results['anomaly_dates'])
plt.show()

実行結果の確認

正常に実行されると、以下のような出力が表示され、グラフウィンドウが開く。数値と日付は実行時点のデータにより異なる。

グラフには、青線で実績値、破線で予測値、赤点で異常値として検出された日の株価が表示される。グラフウィンドウを閉じるとプログラムが終了する。

3. 地理空間データの活用

以下のPythonプログラムは、国土数値情報ダウンロードサービスから取得した地理空間データを読み込み、観測地点との空間結合（複数のレイヤの地理データを位置関係に基づいて結合する処理）を行う実装例である。

データの準備

国土数値情報のデータは、以下の手順で事前にダウンロードする。

1. 国土数値情報ダウンロードサービス（https://nlftp.mlit.go.jp/ksj/）にアクセスする
2. 「行政区域」を選択する
3. 対象年度（例：令和7年）と都道府県（例：福岡県）を選択し、シェープファイル（GISで広く使用されるベクタデータ形式）形式でダウンロードする
4. ダウンロードしたZIPファイルを解凍する
5. 解凍したフォルダをPythonプログラムと同じディレクトリに配置する

ファイル配置の例を以下に示す。

作業ディレクトリ/
├── spatial_analysis.py          （Pythonプログラム）
└─N03-20250101_40_GML/          （解凍したフォルダ）
    ├─KS-META-N03-20250101_40.xml メタデータ
    ├── N03-20250101_40.xml XML
    ├── N03-20250101_40.geojson  GEOJSON
    ├── N03-20250101_40.cpg  CPG
    ├── N03-20250101_40.shp      （シェープファイル本体）
    ├── N03-20250101_40.dbf      （属性データ）
    ├── N03-20250101_40.shx      （インデックス）
    └── N03-20250101_40.prj      （座標系情報）

プログラム

import geopandas as gpd
import pandas as pd
import matplotlib.pyplot as plt
import japanize_matplotlib
from shapely.geometry import Point

def analyze_spatial_data(shp_path):
    """
    地理空間データの分析を行う
    shp_path: 国土数値情報からダウンロードしたシェープファイルのパス
              例: "N03-20250101_40_GML/N03-20250101_40.shp"
    """
    points_data = {
        'latitude': [33.5902, 33.5895, 33.5932],  # サンプルポイント（福岡市内）
        'longitude': [130.4017, 130.4021, 130.3992],
        'name': ['地点A', '地点B', '地点C']
    }

    # シェープファイルの読み込み
    gdf = gpd.read_file(shp_path)

    # 福岡市のデータのみを抽出
    fukuoka_gdf = gdf[gdf['N03_004'] == '福岡市']

    # 点データの作成
    points_df = pd.DataFrame(points_data)
    points_gdf = gpd.GeoDataFrame(
        points_df,
        geometry=[Point(xy) for xy in zip(points_df['longitude'], points_df['latitude'])],
        crs="EPSG:4326"
    )

    # 座標参照系を統一
    if fukuoka_gdf.crs != points_gdf.crs:
        points_gdf = points_gdf.to_crs(fukuoka_gdf.crs)

    # 空間結合（predicateパラメータを使用）
    joined = gpd.sjoin(points_gdf, fukuoka_gdf, how='left', predicate='within')

    # 地図プロット
    fig, ax = plt.subplots(figsize=(10, 10))
    fukuoka_gdf.plot(ax=ax, alpha=0.5)
    points_gdf.plot(ax=ax, color='red', markersize=50)
    plt.title('福岡市地図と観測地点')

    return joined

このプログラムは、国土数値情報ダウンロードサービスから取得した福岡市の地理空間データを読み込み、GeoPandasライブラリを活用して空間分析を実行する。観測地点データとの空間結合処理や視覚化機能を提供し、地理情報システムの基本的なワークフローを実装している。

実行手順

上記の関数定義と以下の実行コードを同じPythonファイル（例：spatial_analysis.py）に保存し、実行する。shp_pathは実際のファイルパスに合わせて変更する。

result = analyze_spatial_data("N03-20250101_40_GML/N03-20250101_40.shp")
print(result[['name', 'N03_004']])
plt.show()

実行結果の確認

正常に実行されると、以下のような出力が表示され、地図ウィンドウが開く。

N03_004列に「福岡市」と表示されていれば、各地点が福岡市の行政区域内に位置していることが空間結合により判定された結果である。地図ウィンドウには、福岡市の行政区域が薄い色で、観測地点が赤い点で表示される。地図ウィンドウを閉じるとプログラムが終了する。

まとめ

本文書では、リレーショナルデータベースとPythonを組み合わせた応用例を学んだ。RDBを活用することで、以下の利点が得られる。

• SQLによるデータ操作が可能であり、検索や集計を効率的に実行できる
• Pythonなどのプログラミング言語との連携が容易であり、データの取得から分析までを自動化できる
• 多様なデータ形式への対応が可能である
  • 為替・株価などの金融データ（時系列での蓄積と統計分析）
  • 時系列データ（ARIMAモデルによる予測、異常値検出）
  • 空間データ（地理情報システムとの連携、空間結合による分析）