TestForge | 📊 Plogger ✍️ Blog 📚 Docs
TestForge Blog

Python AI 가이드 · 03

데이터 처리

NumPy, Pandas, Matplotlib. AI 모델에 데이터를 넣기 전 탐색·변환·정제하는 전체 흐름입니다.

1. NumPy — 수치 배열 연산

import numpy as np

a    = np.array([1, 2, 3, 4, 5], dtype=np.float32)
zeros = np.zeros((3, 4))
rng  = np.random.default_rng(42)
data = rng.normal(0, 1, (100, 5))

# 벡터화 연산
prices  = np.array([50000, 52000, 48000, 55000])
returns = np.diff(prices) / prices[:-1]
print(returns.mean(), returns.std())

# 인덱싱
data[0]         # 첫 행
data[:, 2]      # 3번째 열
data[data > 0]  # 양수만

# 브로드캐스팅
row_mean   = data.mean(axis=1, keepdims=True)
normalized = data - row_mean

2. Pandas — 표 형태 데이터

import pandas as pd

df = pd.read_csv("stocks.csv", parse_dates=["date"])

# 탐색
df.shape ; df.dtypes ; df.describe() ; df.isnull().sum()

# 선택
df["close"]
df[["code", "close", "volume"]]
df[df["volume"] > 1_000_000]
df.loc[df["code"] == "005930"]

# 변환
df["change_rate"] = df["close"].pct_change() * 100
df["ma5"]         = df["close"].rolling(5).mean()
df["ma20"]        = df["close"].rolling(20).mean()

# 그룹 집계
daily = (
    df.groupby("date")
    .agg({"volume": "sum", "close": "last"})
    .reset_index()
)

df.fillna(method="ffill", inplace=True)

# 외인 수급 5일 누적

df["fp_net_cum5"] = (
    df.groupby("code")["fp_net"]
    .rolling(5, min_periods=1).sum()
    .reset_index(level=0, drop=True)
)

top10 = (
    df[df["date"] == df["date"].max()]
    .nlargest(10, "fp_net_cum5")[["code", "name", "fp_net_cum5"]]
)
print(top10.to_string(index=False))

3. Matplotlib — 시각화

import matplotlib.pyplot as plt
import matplotlib.dates as mdates

fig, axes = plt.subplots(2, 1, figsize=(12, 8), sharex=True)

ax1 = axes[0]
ax1.plot(df["date"], df["close"], color="#059669", label="종가")
ax1.plot(df["date"], df["ma5"],   color="#f59e0b", label="MA5")
ax1.plot(df["date"], df["ma20"],  color="#6366f1", label="MA20")
ax1.legend() ; ax1.grid(alpha=0.3)

ax2 = axes[1]
colors = ["#ef4444" if c < 0 else "#3b82f6" for c in df["change_rate"]]
ax2.bar(df["date"], df["volume"], color=colors, alpha=0.7)
ax2.xaxis.set_major_formatter(mdates.DateFormatter("%m/%d"))

plt.tight_layout()
plt.savefig("chart.png", dpi=150, bbox_inches="tight")

4. AI 모델 입력 전처리

from sklearn.preprocessing import StandardScaler
import numpy as np

features = ["change_rate", "volume_ratio", "buy_pressure", "vwap_dev"]
X = df[features].values

valid    = ~np.isnan(X).any(axis=1)
X        = X[valid]
X_scaled = StandardScaler().fit_transform(X)

split = int(len(X_scaled) * 0.8)
X_train, X_val = X_scaled[:split], X_scaled[split:]
print(f"훈련: {X_train.shape}, 검증: {X_val.shape}")