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}")