1분봉 ?
1년동안 코스피 역대급 떡상이라 별로 도움 안될듯한데요 1분봉 ? 1년동안 코스피 역대급 떡상이라 별로 도움 안될듯한데요

원래 10년치 였었거든요 ㅠㅠ.... 원래 10년치 였었거든요 ㅠㅠ....

대신증권 분봉 2년치 조회되지않나요? 대신증권 분봉 2년치 조회되지않나요?

네 맞습니다~ 네 맞습니다~

제가 말씀드린대로 증권사 API로 하시는게 젤 편할거예요 제가 말씀드린대로 증권사 API로 하시는게 젤 편할거예요

# src/toss_api.py

import requests

from .config import (
    TOSS_CLIENT_ID,
    TOSS_CLIENT_SECRET,
    TOSS_BASE_URL,
    validate_config,
)


class TossAPI:

    def __init__(self):
        validate_config()

        self.base_url = TOSS_BASE_URL
        self.session = requests.Session()

    # -----------------------------------------
    # Toss OAuth 인증
    # -----------------------------------------

    def _request_new_token(self):

        url = f"{self.base_url}/oauth2/token"

        data = {
            "grant_type": "client_credentials",
            "client_id": TOSS_CLIENT_ID,
            "client_secret": TOSS_CLIENT_SECRET,
        }

        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }

        response = self.session.post(
            url,
            data=data,
            headers=headers,
            timeout=10,
        )

        result = response.json()

        return result["access_token"]

    # -----------------------------------------
    # 공통 GET 요청
    # -----------------------------------------

    def get(self, endpoint, params=None):

        self.ensure_token()

        url = f"{self.base_url}{endpoint}"

        response = self.session.get(
            url,
            params=params,
            timeout=10,
        )

        return response.json()

    # -----------------------------------------
    # 전체 종목 목록
    # -----------------------------------------

    def get_all_stocks(self, market, status="ACTIVE"):

        return self.get(
            "/api/v1/stocks/all",
            params={
                "market": market,
                "status": status,
            },
        )

    # -----------------------------------------
    # 캔들 데이터
    # -----------------------------------------

    def get_candles(
        self,
        symbol,
        interval="1d",
        count=30,
        before=None,
        adjusted=True,
    ):

        params = {
            "symbol": symbol,
            "interval": interval,
            "count": count,
            "adjusted": str(adjusted).lower(),
        }

        if before is not None:
            params["before"] = before

        return self.get(
            "/api/v1/candles",
            params=params,
        ) # src/toss_api.py import requests from .config import ( TOSS_CLIENT_ID, TOSS_CLIENT_SECRET, TOSS_BASE_URL, validate_config, ) class TossAPI: def __init__(self): validate_config() self.base_url = TOSS_BASE_URL self.session = requests.Session() # ----------------------------------------- # Toss OAuth 인증 # ----------------------------------------- def _request_new_token(self): url = f"{self.base_url}/oauth2/token" data = { "grant_type": "client_credentials", "client_id": TOSS_CLIENT_ID, "client_secret": TOSS_CLIENT_SECRET, } headers = { "Content-Type": "application/x-www-form-urlencoded" } response = self.session.post( url, data=data, headers=headers, timeout=10, ) result = response.json() return result["access_token"] # ----------------------------------------- # 공통 GET 요청 # ----------------------------------------- def get(self, endpoint, params=None): self.ensure_token() url = f"{self.base_url}{endpoint}" response = self.session.get( url, params=params, timeout=10, ) return response.json() # ----------------------------------------- # 전체 종목 목록 # ----------------------------------------- def get_all_stocks(self, market, status="ACTIVE"): return self.get( "/api/v1/stocks/all", params={ "market": market, "status": status, }, ) # ----------------------------------------- # 캔들 데이터 # ----------------------------------------- def get_candles( self, symbol, interval="1d", count=30, before=None, adjusted=True, ): params = { "symbol": symbol, "interval": interval, "count": count, "adjusted": str(adjusted).lower(), } if before is not None: params["before"] = before return self.get( "/api/v1/candles", params=params, )

일단 이게 API 수집에 필요한 핵심 코드입니다. 일단 이게 API 수집에 필요한 핵심 코드입니다.

증권사 API로 1분봉 RAW 데이터를 받는 구조는 크게 2개의 Py로 나눠서 생각하시면 됩니다.

```python
# 1) research/large_n_common_raw_acquisition.py
# 목적: "어떤 종목의 1분봉 데이터를 수집할지" 정하는 파일

def eligible_common_stock_universe(payload):
    symbols = []

    for stock in payload:
        # 예: KOSPI + 일반 주식 + 보통주만 통과
        if stock["market"] != "KOSPI":
            continue
        if stock["securityType"] != "STOCK":
            continue
        if stock["isCommonShare"] is not True:
            continue

        symbols.append(stock["code"])

    return symbols


def main():
    # 위의 종목 선정 규칙을 등록한 뒤
    # 실제 1분봉 RAW 수집 프로그램을 실행합니다.
    set_eligible_universe(eligible_common_stock_universe)
    raw_main()
```

이 파일의 목적은 간단합니다.

증권사에 등록된 여러 금융상품 중에서 실제로 1분봉 데이터를 받을 종목만 골라주는 파일입니다.

예를 들어 KOSPI 보통주만 수집하려면 ETF, ETN같은 것들은 제외하고 필요한 종목코드만 목록으로 만듭니다.

---

```python
# 2) research/historical_replay_raw_acquisition_v2.py
# 목적: 선정된 종목의 1분봉 데이터를 실제 API에서 받아서 저장(제가 사용하는 토스로 예를 들겠습니다.)

api = TossAPI()

# Toss에서 전체 종목 목록을 받습니다.
stocks = api.get_all_stocks(
    "KOSPI",
    status="ACTIVE"
)

for symbol in symbols:

    # 실제 1분봉 데이터 요청
    data = api.get_candles(
        symbol=symbol,
        interval="1m",  # 1분봉
        count=200,      # 한 번에 최대 200개 요청(여러 갯수를 테스트해봤는데 200개가 api차단 걱정 없이 젤 적당합니다.)
        before=before,  # 이 시각 이전 데이터를 요청
        adjusted=True
    )

    # Toss에서 받은 JSON 응답을 RAW 파일로 그대로 저장합니다.
    save_raw(data)

    # 데이터가 더 있으면 Toss가 돌려준 nextBefore 값을 이용해 그보다 이전 1분봉 200개를 다시 요청합니다.
    before = data["result"]["nextBefore"]
```

이 파일의 목적은 선정된 종목을 하나씩 Toss API에 요청해서 1분봉 데이터를 받고, 받은 JSON 원본을 그대로 저장하는 실제 수집 구조예요.

1분봉 데이터는 양이 많기 때문에 한 번 요청으로 전부 받을 수 없을겁니다.

따라서 밑에처럼 하시면 될 거 같아요.

```text
최근 1분봉 200개 요청
        ↓
Toss가 데이터 + nextBefore 반환
        ↓
nextBefore 이전의 1분봉 200개 요청
        ↓
또 nextBefore 반환
        ↓
이전 200개 요청
        ↓
계속 반복
        ↓
원하는 시작 날짜까지 도달하면 종료
```

예를 들어 삼성전자라면

```text
삼성전자 005930
      ↓
Toss API /candles
      ↓
1분봉 200개 수신
      ↓
page_001.raw.json 저장
      ↓
nextBefore 확인
      ↓
이전 1분봉 200개 수신
      ↓
page_002.raw.json 저장
      ↓
nextBefore 확인
      ↓
계속 반복
```

그리고 중간에 이더넷 연결이 끊기거나 프로세스가 강제종료되어도 처음부터 다시 받지 않도록:

```text
checkpoint.json
```

파일에

```text
어느 종목까지 받았는지
몇 페이지까지 받았는지
마지막 nextBefore가 무엇인지
몇 개의 데이터를 받았는지
```

등을 기록해 두고, 다시 실행하면 그 지점부터 이어받는 구조로 만들 수 있습니다.

====================

전체 구조를 아주 간단하게 정리하면 다음과 같습니다.

```text
PowerShell
    ↓
python -m research.large_n_common_raw_acquisition
    ↓
large_n_common_raw_acquisition.py
    ↓
수집할 종목 선정
    ↓
historical_replay_raw_acquisition_v2.py
    ↓
Toss API에 1분봉 요청
    ↓
종목별 200개씩 수신
    ↓
nextBefore로 과거 방향 Pagination
    ↓
RAW JSON 저장
    ↓
checkpoint 저장
    ↓
중단되면 이어받기
```

각 파일의 역할은 다음과 같습니다.

```text
large_n_common_raw_acquisition.py
= 어떤 종목을 받을지 결정

historical_replay_raw_acquisition_v2.py(v2는 제가 수정한 py파일 버전이니 신경 안 쓰셔도 됩니다.)
= 선정된 종목의 1분봉 데이터를 반복해서 받아 저장

src/toss_api.py
= Toss 서버와 실제 HTTP 통신을 담당
```

그리고 가장 중요한 부분은 다음입니다.

```python
interval="1m"
```

기존 일봉 수집 코드에서는:

```python
interval="1d"
```

처럼 되어 있는데, 1분봉을,즉  `1m`을 실제 지원한다면 이를 1분 단위로 요청하는 구조로 변경하면 됩니다.
다만 아래 5가지는 확인하셔야해요.
```text
1. Toss API가 interval="1m"을 실제 지원하는지
2. 1분봉을 과거 어느 시점까지 제공하는지
3. 한 요청당 최대 count가 몇 개인지
4. API 호출 제한(Rate Limit)이 있는지
5. nextBefore Pagination이 1분봉에서도 동일하게 동작하는지
```

1분봉은 일봉보다 데이터 양이 압도적으로 많습니다.

예를 들어 장중 1거래일에 약 390개의 1분봉이 있다고 가정하면:

```text
1종목 × 1일
= 390 rows

1종목 × 1년(약 250거래일)
= 97,500 rows

800종목 × 1년
= 7,800만 rows
```

정도가 될 수 있습니다.

그래서 Pagination, API 호출제한, SSD용량, 오류 발생 시 재개, symbol별 checkpoint 구조까지 같이 설계하는 것이 중요합니다.

대충 어떤 구조인지는 이해하시겠지요? 증권사 API로 1분봉 RAW 데이터를 받는 구조는 크게 2개의 Py로 나눠서 생각하시면 됩니다. ```python # 1) research/large_n_common_raw_acquisition.py # 목적: "어떤 종목의 1분봉 데이터를 수집할지" 정하는 파일 def eligible_common_stock_universe(payload): symbols = [] for stock in payload: # 예: KOSPI + 일반 주식 + 보통주만 통과 if stock["market"] != "KOSPI": continue if stock["securityType"] != "STOCK": continue if stock["isCommonShare"] is not True: continue symbols.append(stock["code"]) return symbols def main(): # 위의 종목 선정 규칙을 등록한 뒤 # 실제 1분봉 RAW 수집 프로그램을 실행합니다. set_eligible_universe(eligible_common_stock_universe) raw_main() ``` 이 파일의 목적은 간단합니다. 증권사에 등록된 여러 금융상품 중에서 실제로 1분봉 데이터를 받을 종목만 골라주는 파일입니다. 예를 들어 KOSPI 보통주만 수집하려면 ETF, ETN같은 것들은 제외하고 필요한 종목코드만 목록으로 만듭니다. --- ```python # 2) research/historical_replay_raw_acquisition_v2.py # 목적: 선정된 종목의 1분봉 데이터를 실제 API에서 받아서 저장(제가 사용하는 토스로 예를 들겠습니다.) api = TossAPI() # Toss에서 전체 종목 목록을 받습니다. stocks = api.get_all_stocks( "KOSPI", status="ACTIVE" ) for symbol in symbols: # 실제 1분봉 데이터 요청 data = api.get_candles( symbol=symbol, interval="1m", # 1분봉 count=200, # 한 번에 최대 200개 요청(여러 갯수를 테스트해봤는데 200개가 api차단 걱정 없이 젤 적당합니다.) before=before, # 이 시각 이전 데이터를 요청 adjusted=True ) # Toss에서 받은 JSON 응답을 RAW 파일로 그대로 저장합니다. save_raw(data) # 데이터가 더 있으면 Toss가 돌려준 nextBefore 값을 이용해 그보다 이전 1분봉 200개를 다시 요청합니다. before = data["result"]["nextBefore"] ``` 이 파일의 목적은 선정된 종목을 하나씩 Toss API에 요청해서 1분봉 데이터를 받고, 받은 JSON 원본을 그대로 저장하는 실제 수집 구조예요. 1분봉 데이터는 양이 많기 때문에 한 번 요청으로 전부 받을 수 없을겁니다. 따라서 밑에처럼 하시면 될 거 같아요. ```text 최근 1분봉 200개 요청 ↓ Toss가 데이터 + nextBefore 반환 ↓ nextBefore 이전의 1분봉 200개 요청 ↓ 또 nextBefore 반환 ↓ 이전 200개 요청 ↓ 계속 반복 ↓ 원하는 시작 날짜까지 도달하면 종료 ``` 예를 들어 삼성전자라면 ```text 삼성전자 005930 ↓ Toss API /candles ↓ 1분봉 200개 수신 ↓ page_001.raw.json 저장 ↓ nextBefore 확인 ↓ 이전 1분봉 200개 수신 ↓ page_002.raw.json 저장 ↓ nextBefore 확인 ↓ 계속 반복 ``` 그리고 중간에 이더넷 연결이 끊기거나 프로세스가 강제종료되어도 처음부터 다시 받지 않도록: ```text checkpoint.json ``` 파일에 ```text 어느 종목까지 받았는지 몇 페이지까지 받았는지 마지막 nextBefore가 무엇인지 몇 개의 데이터를 받았는지 ``` 등을 기록해 두고, 다시 실행하면 그 지점부터 이어받는 구조로 만들 수 있습니다. ==================== 전체 구조를 아주 간단하게 정리하면 다음과 같습니다. ```text PowerShell ↓ python -m research.large_n_common_raw_acquisition ↓ large_n_common_raw_acquisition.py ↓ 수집할 종목 선정 ↓ historical_replay_raw_acquisition_v2.py ↓ Toss API에 1분봉 요청 ↓ 종목별 200개씩 수신 ↓ nextBefore로 과거 방향 Pagination ↓ RAW JSON 저장 ↓ checkpoint 저장 ↓ 중단되면 이어받기 ``` 각 파일의 역할은 다음과 같습니다. ```text large_n_common_raw_acquisition.py = 어떤 종목을 받을지 결정 historical_replay_raw_acquisition_v2.py(v2는 제가 수정한 py파일 버전이니 신경 안 쓰셔도 됩니다.) = 선정된 종목의 1분봉 데이터를 반복해서 받아 저장 src/toss_api.py = Toss 서버와 실제 HTTP 통신을 담당 ``` 그리고 가장 중요한 부분은 다음입니다. ```python interval="1m" ``` 기존 일봉 수집 코드에서는: ```python interval="1d" ``` 처럼 되어 있는데, 1분봉을,즉 `1m`을 실제 지원한다면 이를 1분 단위로 요청하는 구조로 변경하면 됩니다. 다만 아래 5가지는 확인하셔야해요. ```text 1. Toss API가 interval="1m"을 실제 지원하는지 2. 1분봉을 과거 어느 시점까지 제공하는지 3. 한 요청당 최대 count가 몇 개인지 4. API 호출 제한(Rate Limit)이 있는지 5. nextBefore Pagination이 1분봉에서도 동일하게 동작하는지 ``` 1분봉은 일봉보다 데이터 양이 압도적으로 많습니다. 예를 들어 장중 1거래일에 약 390개의 1분봉이 있다고 가정하면: ```text 1종목 × 1일 = 390 rows 1종목 × 1년(약 250거래일) = 97,500 rows 800종목 × 1년 = 7,800만 rows ``` 정도가 될 수 있습니다. 그래서 Pagination, API 호출제한, SSD용량, 오류 발생 시 재개, symbol별 checkpoint 구조까지 같이 설계하는 것이 중요합니다. 대충 어떤 구조인지는 이해하시겠지요?

저 같은 경우엔 그래서 LARGE-N 연구로 805 symbol에 일봉으로 8년치 정도의 데이터를 가지고 있습니다. 저도 분봉으로 데이터를 쌓으려고 했던 적이 있었는데, ssd용량이,,,어머어마하게 들어갑니다. 지금 일봉으로만 하는데도 100GB가 넘는걸요. 로컬 역량이 되시면 한번 시도는 해보심이 좋을 거 같네요. 저는 일단 전문적인 프로그램도 아니고, 개인 취미용이라 이렇게 쓰고 있는데, 도움이 되셨으면 좋겠네요. 저 같은 경우엔 그래서 LARGE-N 연구로 805 symbol에 일봉으로 8년치 정도의 데이터를 가지고 있습니다. 저도 분봉으로 데이터를 쌓으려고 했던 적이 있었는데, ssd용량이,,,어머어마하게 들어갑니다. 지금 일봉으로만 하는데도 100GB가 넘는걸요. 로컬 역량이 되시면 한번 시도는 해보심이 좋을 거 같네요. 저는 일단 전문적인 프로그램도 아니고, 개인 취미용이라 이렇게 쓰고 있는데, 도움이 되셨으면 좋겠네요.

지금은 5,20,60거래일 후 실제 결과와 비교하는 딥러닝 예측 모델을 준비하고있습니다.
해당 모델(엔진)에서 무수히 많은 t 들을 예측해서 그 예측값이 미래의 t와 일치하는지 분석하고, 의미있는 지표들은 뭐 였는지, 어떤부분에서 재현성이 젤 좋은지,  성숙된 t를 넣었을 시 또 예측을 어떻게하는지 이런식으로 굴려보고 있어요.
그렇게 해서 scoring이 좀 안정되면(현재 목표는 85%입니다.) 기존에 사용하는 edge scoring에 접목해서 써보려고요.  (근데 이거도 해외 이슈들이랑 섞이면 또 bias가 너무 많이 생겨요 ㅜㅜ)
AI매매 프로그램이면 저처럼 이런식으로 접근해보셔도 좋을 거 같아요.
저야 scoring에서 끝나지만 이 신뢰도 있는 score로 매매 판단까지 맡겨보면 재밌을 거 같긴 하네요.
저도 이번 프로젝트 끝나면 AI매매로 한번 해보려고해요.
한 100만원 정도 증권계좌에 넣어두고 자동으로 위의 scoring engine의 rank + claude나 open ai 같은 api로 실시간 종목 뉴스, 정보 같은걸 섞어서 판별하게 한 뒤 매매까지 이어지게 만들어보면 심심할때마다 어떻게 하고있나~ 들여다 보면 재밌을 거 같긴 합니다. 지금은 5,20,60거래일 후 실제 결과와 비교하는 딥러닝 예측 모델을 준비하고있습니다. 해당 모델(엔진)에서 무수히 많은 t 들을 예측해서 그 예측값이 미래의 t와 일치하는지 분석하고, 의미있는 지표들은 뭐 였는지, 어떤부분에서 재현성이 젤 좋은지, 성숙된 t를 넣었을 시 또 예측을 어떻게하는지 이런식으로 굴려보고 있어요. 그렇게 해서 scoring이 좀 안정되면(현재 목표는 85%입니다.) 기존에 사용하는 edge scoring에 접목해서 써보려고요. (근데 이거도 해외 이슈들이랑 섞이면 또 bias가 너무 많이 생겨요 ㅜㅜ) AI매매 프로그램이면 저처럼 이런식으로 접근해보셔도 좋을 거 같아요. 저야 scoring에서 끝나지만 이 신뢰도 있는 score로 매매 판단까지 맡겨보면 재밌을 거 같긴 하네요. 저도 이번 프로젝트 끝나면 AI매매로 한번 해보려고해요. 한 100만원 정도 증권계좌에 넣어두고 자동으로 위의 scoring engine의 rank + claude나 open ai 같은 api로 실시간 종목 뉴스, 정보 같은걸 섞어서 판별하게 한 뒤 매매까지 이어지게 만들어보면 심심할때마다 어떻게 하고있나~ 들여다 보면 재밌을 거 같긴 합니다.

감사합니다.. 저랑 목표가 같으시네요!
저도 갠적으로 AI매매 준비중이거든요.
사실 주식은 1도 모르고요. 바이브코딩으로 만들고 있습니다.
클로드+chatgpt+그록 이렇게 3개로 번갈아 가면서 디버깅하면서 만들고 있습니다.
chatgpt가 생성하고, 그록+클로드가 반박+지적+개선 사항을 주면 gpt가 다시 만들고.. 다시 그록+클로드가 반박,개선,추가..형식으로요.
그래서 제가 잘 모르는 주식의 함정을 3개로 돌려막기 하고 있습니다.
중요한게 토스api로 8년치 분봉 쌓기가 될까요? 감사합니다.. 저랑 목표가 같으시네요! 저도 갠적으로 AI매매 준비중이거든요. 사실 주식은 1도 모르고요. 바이브코딩으로 만들고 있습니다. 클로드+chatgpt+그록 이렇게 3개로 번갈아 가면서 디버깅하면서 만들고 있습니다. chatgpt가 생성하고, 그록+클로드가 반박+지적+개선 사항을 주면 gpt가 다시 만들고.. 다시 그록+클로드가 반박,개선,추가..형식으로요. 그래서 제가 잘 모르는 주식의 함정을 3개로 돌려막기 하고 있습니다. 중요한게 토스api로 8년치 분봉 쌓기가 될까요?

오~ AI를 정확히 잘 사용하시네요. 저도 GPT, Claude 사용하면서 가끔 제미나이도 쓰고 그럽니다.
주된 개발은 GPT codex로 진행하고요. 요즘 사람보다 똑똑한거같아요 AI들이.
네. 가능은 합니다. 저는 시도하다가 2년치 정도에서 용량부족으로 끊기길래 일봉으로 갈아탔지만요. 오~ AI를 정확히 잘 사용하시네요. 저도 GPT, Claude 사용하면서 가끔 제미나이도 쓰고 그럽니다. 주된 개발은 GPT codex로 진행하고요. 요즘 사람보다 똑똑한거같아요 AI들이. 네. 가능은 합니다. 저는 시도하다가 2년치 정도에서 용량부족으로 끊기길래 일봉으로 갈아탔지만요.

상장폐지 종목까지 싹 다 구하려면 데이터 정제만 몇 달 걸릴 듯.. 상장폐지 종목까지 싹 다 구하려면 데이터 정제만 몇 달 걸릴 듯..