수집한 데이터를 MySQL에 저장하기

CREATE DATABASE IF NOT EXISTS auto_trading_lab
  CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;

USE auto_trading_lab;

CREATE TABLE IF NOT EXISTS universe_snapshots (
  snapshot_date DATE NOT NULL,
  symbol CHAR(6) NOT NULL,
  name VARCHAR(100) NOT NULL,
  source VARCHAR(500) NOT NULL,
  PRIMARY KEY (snapshot_date, symbol)
) ENGINE=InnoDB;

CREATE TABLE IF NOT EXISTS daily_prices (
  symbol CHAR(6) NOT NULL,
  trade_date DATE NOT NULL,
  open DECIMAL(18,4) NOT NULL,
  high DECIMAL(18,4) NOT NULL,
  low DECIMAL(18,4) NOT NULL,
  close DECIMAL(18,4) NOT NULL,
  volume BIGINT UNSIGNED NOT NULL,
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
    ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (symbol, trade_date),
  CONSTRAINT chk_daily_prices_positive
    CHECK (open > 0 AND high > 0 AND low > 0 AND close > 0),
  CONSTRAINT chk_daily_prices_range
    CHECK (high >= open AND high >= close AND high >= low
           AND low <= open AND low <= close)
) ENGINE=InnoDB;

mysql --host=127.0.0.1 --port=3306 --user=lab_writer --password auto_trading_lab

import argparse
from decimal import Decimal
from pathlib import Path

import mysql.connector
import pandas as pd

from src.kis_readonly import load_config

REQUIRED = ["symbol", "trade_date", "open", "high", "low", "close", "volume"]
SQL = """
INSERT INTO daily_prices
  (symbol, trade_date, open, high, low, close, volume)
VALUES (%s, %s, %s, %s, %s, %s, %s) AS incoming
ON DUPLICATE KEY UPDATE
  open = incoming.open, high = incoming.high, low = incoming.low,
  close = incoming.close, volume = incoming.volume
"""


def read_rows(path):
    frame = pd.read_csv(path, dtype="string", encoding="utf-8-sig")
    if set(REQUIRED) - set(frame.columns) or frame.empty:
        raise ValueError("The CSV is empty or missing required columns")
    if frame[REQUIRED].isna().any().any():
        raise ValueError("The CSV contains missing values")
    if not frame.symbol.str.fullmatch(r"[A-Z0-9]{6}").all():
        raise ValueError("Invalid stock symbol")
    frame["trade_date"] = pd.to_datetime(
        frame.trade_date, format="%Y-%m-%d", errors="raise"
    ).dt.date
    if frame.duplicated(["symbol", "trade_date"]).any():
        raise ValueError("Duplicate symbol/date in the input")
    rows = []
    for record in frame[REQUIRED].itertuples(index=False, name=None):
        symbol, trade_date, *numbers = record
        prices = [Decimal(value) for value in numbers[:4]]
        volume_value = Decimal(numbers[4])
        if not all(p.is_finite() and p > 0 for p in prices):
            raise ValueError("Invalid price")
        if any(p != p.quantize(Decimal("0.0001")) for p in prices):
            raise ValueError("A price exceeds four decimal places")
        if (not volume_value.is_finite() or volume_value < 0
                or volume_value != volume_value.to_integral_value()):
            raise ValueError("Invalid volume")
        open_, high, low, close = prices
        if high < max(open_, low, close) or low > min(open_, high, close):
            raise ValueError("Invalid OHLC range")
        rows.append((symbol, trade_date, *prices, int(volume_value)))
    return rows


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("csv", type=Path)
    args = parser.parse_args()
    rows = read_rows(args.csv)
    config = load_config()
    required = ["MYSQL_HOST", "MYSQL_USER", "MYSQL_PASSWORD", "MYSQL_DATABASE"]
    if any(not config.get(key) for key in required):
        raise ValueError("MySQL settings are incomplete")
    connection = mysql.connector.connect(
        host=config["MYSQL_HOST"],
        port=int(config.get("MYSQL_PORT", "3306")),
        user=config["MYSQL_USER"],
        password=config["MYSQL_PASSWORD"],
        database=config["MYSQL_DATABASE"],
        charset="utf8mb4",
        connection_timeout=10,
        autocommit=False,
    )
    cursor = None
    try:
        cursor = connection.cursor()
        for row in rows:
            cursor.execute(SQL, row)
        connection.commit()
        print(f"committed input_rows={len(rows)} file={args.csv.name}")
    except Exception:
        connection.rollback()
        raise
    finally:
        if cursor is not None:
            cursor.close()
        connection.close()


if __name__ == "__main__":
    main()

.\.venv\Scripts\python.exe -m py_compile src/load_prices.py
.\.venv\Scripts\python.exe -m src.load_prices outputs/daily/005930_2026-01-01_2026-10-16.csv

SELECT symbol, COUNT(*) AS rows_count,
       MIN(trade_date) AS first_date, MAX(trade_date) AS last_date
FROM daily_prices
GROUP BY symbol
ORDER BY symbol;

SELECT symbol, trade_date, open, high, low, close, volume
FROM daily_prices
WHERE symbol = '005930'
ORDER BY trade_date DESC
LIMIT 5;