6 / 12 · AUTO TRADING WIKI

MySQL 데이터로 CCI 신호 계산하기

가격 입력값과 계산 기간을 고정하고 계산 결과와 신호를 기록합니다.

발행: 2026-10-20
수정: 2026-10-20
문서 검토: 2026-10-20
본문 버전: 1.1.2-web.20261020

MySQL에 저장한 일봉을 이용해 CCI를 계산하고, 정해진 경계선을 넘는 날을 신호로 기록한다. 이번 장의 결과는 종목별 20거래일 CCI와 진입 후보 신호다. 이 신호를 주문으로 전환하지 않으며, 다음 장에서 과거 데이터로 동작과 한계를 확인한다.

예제 규칙은 전일 CCI <= -100이고 당일 CCI > -100일 때 신호를 한 번 기록하는 것이다. 이는 계산과 시뮬레이션을 연결하기 위한 학습 규칙이며, 특정 종목이나 매매 방법의 수익성을 추천하는 뜻이 아니다.

1. CCI 계산식 이해하기

CCI는 오늘의 대표가격이 최근 대표가격 평균에서 얼마나 떨어져 있는지를 평균절대편차로 나눈 지표다. 대표가격 TP는 고가·저가·종가의 평균이다. 기간을 20거래일로 정하면 오늘을 포함한 최근 20개 관측값을 한 창으로 계산한다.

TP[t] = (high[t] + low[t] + close[t]) / 3
SMA[t] = 최근 20개 TP의 평균
MD[t] = 최근 20개 각각의 TP와 동일한 SMA[t] 간 절대 차이의 평균
CCI[t] = (TP[t] - SMA[t]) / (0.015 × MD[t])

MD[t]를 구할 때는 그 창 안의 모든 값에서 같은 창의 평균을 뺀다. 각 날짜의 별도 이동평균을 먼저 빼고 그 차이에 다시 이동평균을 적용하면 다른 계산이 된다. CCI의 대표가격·평균절대편차 공식은 Fidelity의 지표 설명을 참고했다.

CCI는 -100과 100 사이에 제한된 지표가 아니다. 범위 밖으로 크게 움직일 수 있고 낮은 값이 계속 유지될 수도 있다. 따라서 CCI <= -100이라는 상태와 -100을 위로 통과했다는 사건을 구분한다. 이번 규칙은 두 번째 사건만 기록한다.

2. 계산 전 데이터와 예외 기준 정하기

5장의 daily_prices에서 symbol, trade_date, open, high, low, close, volume을 읽는다. 최소한 다음 조건을 통과해야 한다.

종목코드는 ASCII 숫자·영문 대문자 6자리이며, 종목코드와 거래일 조합이 중복되지 않는다.
각 종목 안에서 거래일이 오름차순이다.
가격은 유한한 양수이고 OHLC의 범위가 맞는다.
종목이 바뀌면 이동 계산과 전일 비교도 새로 시작한다.
계산 시작일 전에 필요한 준비 기간을 함께 읽는다.

처음 19개 관측값은 20개가 모이지 않았으므로 CCI가 없다. 20개가 모두 같은 대표가격이면 평균절대편차가 0이므로 역시 계산 불가로 둔다. 두 경우 모두 NaN으로 유지하며 DB에서는 NULL로 저장한다. 0으로 바꾸면 실제 중립값처럼 보일 수 있다.

신호에는 전일과 당일의 유효한 CCI가 모두 필요하다. 따라서 처음 유효한 CCI가 생긴 날은 신호를 만들지 않는다. 거래 정지나 누락으로 관측일 간격이 벌어진 경우, 이번 장의 계산은 존재하는 20개 거래 관측값을 사용한다. 데이터 품질 검사는 계산기 밖에서 별도로 유지한다.

3. 종목별 CCI 계산 함수 작성하기

src/calculate_cci.py에 다음 코드를 작성한다. 이동 창의 min_periods=period를 지정해 준비 기간을 명확히 하고, 미래 값이 들어오는 중앙 정렬은 사용하지 않는다. 창과 최소 관측수의 의미는 pandas 공식 rolling 문서를 참고한다.

import argparse
import math
import re

import mysql.connector
import pandas as pd

from src.kis_readonly import ROOT, load_config


def cross_up(values, threshold=-100.0):
    previous = values.shift(1)
    return (previous.le(threshold) & values.gt(threshold)).fillna(False).astype(bool)


def calculate_cci(frame, period=20):
    if period < 2:
        raise ValueError("period must be at least 2")
    required = {"symbol", "trade_date", "open", "high", "low", "close"}
    if required - set(frame.columns) or frame.empty:
        raise ValueError("Prices are empty or missing required columns")
    data = frame.copy()
    if data[["symbol", "trade_date"]].isna().any().any():
        raise ValueError("Missing symbol or trade date")
    data["trade_date"] = pd.to_datetime(data.trade_date, errors="raise")
    if data["trade_date"].isna().any():
        raise ValueError("Missing trade date")
    if data.duplicated(["symbol", "trade_date"]).any():
        raise ValueError("Duplicate symbol/date")
    for column in ["open", "high", "low", "close"]:
        data[column] = pd.to_numeric(data[column], errors="raise").astype(float)
        if not data[column].map(lambda x: math.isfinite(x) and x > 0).all():
            raise ValueError(f"Invalid price in {column}")
    invalid = (
        data.high.lt(data[["open", "low", "close"]].max(axis=1))
        | data.low.gt(data[["open", "high", "close"]].min(axis=1))
    )
    if invalid.any():
        raise ValueError("Invalid OHLC range")
    data = data.sort_values(["symbol", "trade_date"]).reset_index(drop=True)
    results = []
    for symbol, group in data.groupby("symbol", sort=False):
        group = group.copy()
        typical = (group.high + group.low + group.close) / 3.0
        window = typical.rolling(window=period, min_periods=period)
        average = window.mean()
        deviation = window.apply(lambda values: abs(values - values.mean()).mean(), raw=True)
        denominator = (0.015 * deviation).where(deviation.gt(0))
        cci = (typical - average) / denominator
        group["typical_price"] = typical
        group["mean_deviation"] = deviation
        group[f"cci_{period}"] = cci
        group["entry_signal"] = cross_up(cci)
        results.append(group)
    return pd.concat(results, ignore_index=True)


def read_prices(symbols):
    config = load_config()
    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"],
        connection_timeout=10,
    )
    cursor = None
    try:
        cursor = connection.cursor(dictionary=True)
        marks = ",".join(["%s"] * len(symbols))
        sql = (
            "SELECT symbol, trade_date, open, high, low, close, volume "
            f"FROM daily_prices WHERE symbol IN ({marks}) "
            "ORDER BY symbol, trade_date"
        )
        cursor.execute(sql, tuple(symbols))
        return pd.DataFrame(cursor.fetchall())
    finally:
        if cursor is not None:
            cursor.close()
        connection.close()


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--symbols", nargs="+", required=True)
    args = parser.parse_args()
    if any(not re.fullmatch(r"[A-Z0-9]{6}", s) for s in args.symbols):
        parser.error("Use six-character uppercase ASCII stock symbols")
    frame = read_prices(args.symbols)
    if frame.empty or set(args.symbols) - set(frame["symbol"]):
        raise ValueError("One or more requested symbols have no stored prices")
    result = calculate_cci(frame, period=20)
    target = ROOT / "outputs/indicators"
    target.mkdir(parents=True, exist_ok=True)
    result.to_csv(target / "cci_20.csv", index=False, encoding="utf-8-sig")
    summary = result.groupby("symbol").agg(
        rows=("trade_date", "size"),
        valid_cci=("cci_20", "count"),
        signals=("entry_signal", "sum"),
    )
    print(summary.to_string())


if __name__ == "__main__":
    main()

read_prices는 SQL의 자리표시자 개수만 만들고 종목코드 값은 별도로 바인딩한다. CCI 계산 중 다른 종목의 직전 행을 읽지 않도록 그룹 안에서 shift(1)을 수행한다. 이 예제의 집계와 정렬은 pandas 공식 GroupBy 안내의 데이터 분리 원리를 따른다.

4. 인공 데이터로 계산과 신호 경계 확인하기

실제 주가를 보기 전에 손으로 답을 구할 수 있는 자료를 사용한다. 아래 코드는 scripts/check_cci.py로 저장하고 프로젝트 루트에서 모듈로 실행한다. 가짜 가격은 특정 종목의 실제 수익률 자료가 아니다.

import pandas as pd

from src.calculate_cci import calculate_cci, cross_up


def prices(symbol, values):
    return pd.DataFrame({
        "symbol": symbol,
        "trade_date": pd.date_range("2026-01-01", periods=len(values), freq="D"),
        "open": values, "high": values, "low": values, "close": values,
    })


rising = calculate_cci(prices("000001", list(range(1, 21))))
assert rising.cci_20.iloc[:19].isna().all()
assert abs(rising.cci_20.iloc[19] - 126.6666666667) < 1e-8
assert not rising.entry_signal.any()

flat = calculate_cci(prices("000002", [10] * 25))
assert flat.cci_20.isna().all()
assert not flat.entry_signal.any()

mixed = calculate_cci(pd.concat([
    prices("000001", list(range(1, 26))),
    prices("000002", [10] * 25),
]).sample(frac=1, random_state=7))
assert mixed.groupby("symbol").cci_20.count().to_dict() == {"000001": 6, "000002": 0}

values = pd.Series([float("nan"), -120, -100, -99, -80, -110, -90])
assert cross_up(values).tolist() == [False, False, False, True, False, False, True]
print("synthetic CCI checks OK")

.\.venv\Scripts\python.exe -m scripts.check_cci

첫 예제는 평균 10.5, 평균절대편차 5, 마지막 대표가격 20이므로 CCI가 약 126.6667이다. 두 번째는 0으로 나누는 경우를, 세 번째는 정렬과 종목 간 분리를 확인한다. 마지막 예제에서 정확히 -100에 도착한 날은 신호가 없고 -100보다 커진 날에만 신호가 발생한다.

5. 실제 DB 자료로 결과 확인하기

수집·저장이 끝난 종목 하나부터 실행한다. 실제 보유 종목이나 매수 대상으로 선정하는 절차는 아니다.

.\.venv\Scripts\python.exe -m src.calculate_cci --symbols 005930

outputs/indicators/cci_20.csv를 열어 앞부분의 결측값, 날짜 순서, 대표가격과 평균절대편차를 확인한다. rows는 읽은 일봉 수, valid_cci는 계산 가능한 행 수, signals는 경계 통과 횟수다. 신호가 0건이어도 규칙에 맞는 데이터라면 정상 결과다.

특정 검토 기간만 계산할 때는 그 기간 이전 자료를 충분히 읽고, 계산을 마친 다음 보고할 날짜를 자른다. 처음부터 검토 시작일 이후 행만 읽으면 준비 기간이 잘려 앞부분의 신호가 달라진다. 마지막 일봉은 거래일 종료 후 확정된 자료인지도 확인한다.

6. 계산 결과의 저장 키와 버전 정하기

CSV로 계산을 확인한 뒤 DB에 결과를 보관할 수 있다. 다음 스키마는 지표 값과 신호 규칙의 버전을 함께 남기는 예시다.

CREATE TABLE IF NOT EXISTS cci_daily_signals (
  symbol CHAR(6) NOT NULL,
  trade_date DATE NOT NULL,
  period SMALLINT UNSIGNED NOT NULL,
  strategy_version VARCHAR(40) NOT NULL,
  cci_value DOUBLE NULL,
  entry_signal BOOLEAN NOT NULL,
  calculated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (symbol, trade_date, period, strategy_version)
) ENGINE=InnoDB;

이번 버전 이름은 cci20_cross_up_m100_v1로 정한다. cci_20을 cci_value에 저장하고, NaN은 None으로 변환해 SQL의 NULL로 바인딩한다. 같은 키를 재계산하면 갱신하고, 공식이나 신호 규칙이 바뀌면 버전 이름을 바꾼다. 적재의 트랜잭션과 중복 처리 방식은 5장과 같다.

값을 NULL로 저장하는 것과 값이 0인 것은 다르다. DB를 다시 읽을 때도 결측 의미를 보존한다. MySQL 공식 NULL 설명

7. Codex에 계산 검증 요청하기

daily_prices를 읽어 20거래일 CCI와 -100 상향 돌파 신호를 계산해줘.
MD는 같은 창의 평균에 대한 평균절대편차로 계산해줘.
첫 19개 행과 MD=0은 결측, 전일 값이 없으면 신호 False로 유지해줘.
종목별 정렬·그룹 분리·중복 검사를 하고 미래 값을 사용하지 마.
손계산 가능한 상승 자료, 일정한 가격, 섞인 두 종목, 경계값 사례를 검사해줘.
신호는 전일 <= -100 이면서 당일 > -100일 때만 True로 기록해줘.
결과 CSV와 요약을 만들고, 실제 DB 접속 여부와 합성 검사 결과를 구분해줘.
주문이나 성과 주장은 추가하지 말고 계산 근거와 남은 검증 범위를 보고해줘.

8. 오류 대응과 다음 장의 연결

문제	확인할 내용
처음부터 값이 나옴	최소 관측수가 20인지, 결측을 0으로 채우지 않았는지
값이 계속 무한대임	MD=0을 분모에서 제외했는지
종목 전환 때 신호 발생	전일 비교를 종목별로 했는지
차트 서비스와 값이 다름	수정주가·시장·거래일·기간·MD 공식·반올림 시점을 비교한다
예상보다 신호가 적음	상태를 세는지 경계 통과 사건을 세는지 구분한다

완료 기준은 계산식, 준비 기간, MD=0 처리, 상향 돌파 규칙을 코드와 결과로 설명할 수 있는 상태다. 신호가 발생한 날의 종가를 이미 알고 계산했으므로 그 종가에 즉시 체결된 것으로 가정하지 않는다. 다음 장에서는 T일 신호 → 다음 거래일 시가라는 시간 순서를 고정해 시뮬레이션한다.