타이의 노트

[LottoLab] 2. 로또 당첨 번호 수집 및 SQLite 데이터베이스 구축

1. 개요

LottoLab의 첫 번째 개발 단계는 역대 로또 당첨 번호를 수집하고 관리하는 시스템을 구현하는 것이다

앞으로 구현할 통계 분석과 예측 모델은 모두 과거 당첨 번호를 기반으로 작동한다. 따라서 데이터를 안정적으로 수집하고, 중복이나 누락 없이 저장할 수 있는 구조가 우선적으로 필요하다. 이번 개발에서는 Python의 requests 라이브러리를 사용해 동행복권 웹사이트에서 당첨 번호를 가져오고, 수집한 데이터를 SQLite 데이터베이스에 저장하도록 구현한다

또한 프로그램을 실행할 때마다 전체 데이터를 다시 수집하지 않고, 데이터베이스에 저장된 회차를 확인해 필요한 데이터만 추가하는 갱신 기능도 구현한다

이번 개발의 목표

  • 동행복권 웹사이트에서 역대 로또 당첨 번호 수집
  • 수집한 JSON 데이터의 파싱 및 유효성 검증
  • SQLite 데이터베이스 생성 및 데이터 저장
  • 중복 데이터 방지 및 기존 데이터 갱신
  • 최신 회차 확인 및 누락 데이터 자동 수집




2. 로또 당첨 번호 수집

2.1 데이터 출처

로또 당첨 번호는 동행복권 공식 웹사이트에서 제공하는 데이터를 사용한다. 웹사이트 내부에서 사용하는 다음 엔드포인트를 통해 당첨 결과를 JSON 형식으로 받아올 수 있다

https://www.dhlottery.co.kr/lt645/selectPstLt645Info.do

해당 엔드포인트는 개발자를 대상으로 공식 문서가 제공되는 공개 API가 아니므로, 웹사이트 개편에 따라 동작 방식이 변경될 가능성이 있다

2.2 API 요청 방식

API는 srchLtEpsd 매개변수를 이용해 조회할 회차를 지정할 수 있다. 실제 요청을 통해 다음 세 가지 방식을 확인했다

요청 방식동작
매개변수 없음최신 당첨 회차 조회
srchLtEpsd=100100회 당첨 결과 조회
srchLtEpsd=all전체 당첨 결과 조회

Python에서는 다음과 같이 요청할 수 있다

import requests
 
API_URL = "https://www.dhlottery.co.kr/lt645/selectPstLt645Info.do"
 
# 최신 회차 조회
response = requests.get(API_URL)
 
# 특정 회차 조회
response = requests.get(
    API_URL,
    params={"srchLtEpsd": 100}
)
 
# 전체 회차 조회
response = requests.get(
    API_URL,
    params={"srchLtEpsd": "all"}
)

특히 srchLtEpsd=all을 사용하면 역대 당첨 번호를 한 번의 요청으로 수집할 수 있다. 실험 당시 1회부터 1246회까지 총 1,246개의 당첨 결과가 반환되는 것을 확인했다

2.3 API 응답 구조

API는 JSON 형식으로 데이터를 반환한다. 당첨 번호 데이터는 다음 경로에 저장되어 있다

data["data"]["list]

각 회차의 데이터에서 필요한 주요 필드는 다음과 같다

필드의미
ltEpsd회차 번호
ltRflYmd추첨 날짜
tm1WnNo ~ tm6WnNo당첨 번호 6개
bnsWnNo보너스 번호

프로젝트에서는 당첨 번호의 통계적 분석이 목적이므로, 당첨 금액이나 당첨자 수 등의 부가 정보는 현재 저장하지 않는다





3. 당첨 번호 데이터 검증

API에서 받은 데이터를 그대로 데이터베이스에 저장하기보다는, 먼저 데이터가 올바른 형식인지 검사하도록 구현했다

로또 6/45의 당첨 번호는 다음 조건을 만족해야 한다

  • 당첨 번호는 정확히 6개여야 한다
  • 각 번호는 1 이상 45 이하의 정수여야 한다
  • 6개의 당첨 번호는 서로 중복되지 않아야 한다
  • 보너스 번호 역시 1~45 범위에 있어야 한다
  • 보너스 번호는 당첨 번호 6개와 중복되면 안 된다

3.1 당첨 번호 검증 함수

def check_numbers(numbers):
    if len(numbers) != 6:
        return False
 
    for number in numbers:
        if type(number) is not int:
            return False
 
        if not 1 <= number <= 45:
            return False
 
    return len(set(numbers)) == 6

check_numbers()는 당첨 번호 6개를 전달받아 유효한 데이터인지 검사한다

먼저 len(numbers)로 번호 개수를 확인하고, 각 번호의 자료형과 범위를 검사한다. 마지막으로 set()을 이용해 중복 여부를 확인한다

Python의 집합(set)은 중복된 값을 하나만 보관하기 때문에, 집합으로 변환한 뒤에도 원소가 6개라면 모든 번호가 서로 다르다는 의미다

3.2 API 응답 데이터 변환

API 응답에는 프로젝트에서 사용하지 않는 정보도 포함되어 있다

따라서 필요한 필드만 추출하고, 분석 및 저장에 적합한 형태로 변환하는 parse_result() 함수를 구현했다

def parse_result(item):
    try:
        round_number = item["ltEpsd"]
        date = item["ltRflYmd"]
 
        numbers = [
            item[f"tm{i}WnNo"]
            for i in range(1, 7)
        ]
 
        bonus = item["bnsWnNo"]
 
        if type(round_number) is not int or round_number < 1:
            return None
 
        if not isinstance(date, str) or len(date) != 8:
            return None
 
        if not date.isdigit():
            return None
 
        if not check_numbers(numbers):
            return None
 
        if type(bonus) is not int or not 1 <= bonus <= 45:
            return None
 
        if bonus in numbers:
            return None
 
        return {
            "round": round_number,
            "date": date,
            "numbers": numbers,
            "bonus": bonus
        }
 
    except (KeyError, TypeError, ValueError):
        return None

당첨 번호 6개는 리스트 컴프리헨션(List Comprehension)을 이용해 추출했다

numbers = [
    item[f"tm{i}WnNo"]
    for i in range(1, 7)
]

이 코드는 tm1WnNo부터 tm6WnNo까지의 값을 순서대로 읽어 리스트로 만든다

데이터가 검증을 통과하면 다음과 같은 딕셔너리로 변환된다

{
    "round": 100,
    "date": "20041106",
    "numbers": [1, 7, 11, 23, 37, 42],
    "bonus": 6
}

위 값은 반환 구조를 설명하기 위한 예시다

검증에 실패하면 None을 반환하도록 하여 잘못된 데이터를 저장하지 않도록 했다

4. API 요청 함수 구현

앞서 확인한 세 가지 조회 방식을 하나의 함수에서 처리하도록 request_number()를 구현했다

def request_numbers(round_number=None):
    params = {}
 
    if round_number is not None:
        params["srchLtEpsd"] = round_number
 
    try:
        response = requests.get(
            API_URL,
            params=params,
            timeout=30
        )
 
        response.raise_for_status()
 
        data = response.json()
        items = data["data"]["list"]
 
        if not isinstance(items, list):
            raise ValueError("API 응답이 리스트 형식이 아닙니다.")
 
        results = []
 
        for item in items:
            result = parse_result(item)
 
            if result is None:
                raise ValueError("잘못된 당첨 번호 데이터가 있습니다.")
 
            results.append(result)
 
        return results
 
    except (
        requests.RequestException,
        ValueError,
        KeyError,
        TypeError
    ) as error:
        print(f"데이터 요청 실패: {error}")
        return None

이 함수는 round_number 매개변수에 전달하는 값에 따라 동작이 달라진다

request_numbers()       # 최신 회차
request_numbers(100)    # 100회
request_numbers("all")  # 전체 회차

response.raise_for_status()는 HTTP 요청이 실패했을 경우 예외를 발생시킨다. 또한 timeout=30을 지정해 서버에서 응답이 오지 않을 때 프로그램이 무기한 대기하는 상황을 방지했다

API 요청에 실패하거나 응답 데이터가 올바르지 않으면 None을 반환한다. 반면 정상적인 요청에서는 검증을 통과한 회차별 딕셔너리의 리스트를 반환한다





5. SQLite 데이터베이스 구축

5.1 SQLite 선택 이유

수집한 당첨 번호는 이후 여러 분석 과정에서 반복적으로 사용하게 된다

매번 API에 요청해 데이터를 가져오는 방식은 불필요한 네트워크 통신이 발생하고, 외부 서버 상태에 따라 프로그램 실행이 영향을 받을 수 있다. 따라서 데이터를 로컬 데이터베이스에 저장하기로 했다

SQLite는 별도의 데이터베이스 서버를 설치하거나 운영할 필요가 없으며, 하나의 파일로 데이터를 관리할 수 있다. 로또 당첨 번호처럼 데이터 규모가 비교적 작고 구조가 단순한 경우에 적합하다고 판단했다

5.2 데이터베이스 파일 경로

from pathlib import Path
 
DB_PATH = (
    Path(__file__).resolve().parent
    / "data"
    / "lotto.db"
)

__file__은 현재 Python 파일의 경로를 나타낸다

resolve().parent를 이용하면 프로그램 파일이 위치한 디렉터리를 기준으로 데이터베이스 경로를 설정할 수 있다. 따라서 프로그램을 어느 디렉터리에서 실행하더라도 프로젝트 내부의 data/lotto.db 파일을 사용하게 된다

5.3 테이블 생성

def create_database():
    DB_PATH.parent.mkdir(
        parents=True,
        exist_ok=True
    )
 
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS lotto (
                round INTEGER PRIMARY KEY,
                date TEXT NOT NULL,
                num1 INT NOT NULL,
                num2 INT NOT NULL,
                num3 INT NOT NULL,
                num4 INT NOT NULL,
                num5 INT NOT NULL,
                num6 INT NOT NULL,
                bonus INT NOT NULL
            )
        """)

데이터베이스에는 lotto라는 테이블을 생성한다

컬럼자료형설명
roundINT회차 번호, 기본 키
dateTEXT추첨 날짜
num1 ~ num6INT당첨 번호 6개
bonusINT보너스 번호

round를 기본 키로 지정해 같은 회차가 중복 저장되지 않도록 했다

CREATE TABLE IF NOT EXISTS를 사용했으므로 이미 테이블이 존재한다면 새로 생성하지 않는다

5.4 당첨 번호 저장

def save_numbers(results):
    rows = [
        (
            result["round"],
            result["date"],
            *result["numbers"],
            result["bonus"]
        )
        for result in results
    ]
 
    with sqlite3.connect(DB_PATH) as conn:
        cursor = conn.executemany("""
            INSERT INTO lotto
            (round, date, num1, num2, num3,
             num4, num5, num6, bonus)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(round) DO UPDATE SET
                date = excluded.date,
                num1 = excluded.num1,
                num2 = excluded.num2,
                num3 = excluded.num3,
                num4 = excluded.num4,
                num5 = excluded.num5,
                num6 = excluded.num6,
                bonus = excluded.bonus
        """, rows)
 
        return cursor.rowcount

executemany()는 동일한 SQL 문장을 여러 데이터에 반복 실행하는 함수다

이를 사용하면 여러 회차의 당첨 번호를 하나의 데이터베이스 연결에서 처리할 수 있다. 또한 ON CONFLICT(round) DO UPDATE를 사용해 동일한 회차가 존재하는 경우 기존 데이터를 갱신하도록 했다

이 방식은 이미 저장된 회차를 다시 수집하더라도 기본 키 충돌로 인해 프로그램이 중단되지 않는다는 장점이 있다. 다만 기존 데이터를 갱신할 수 있다는 것은 API 응답에 잘못된 값이 포함되면 저장된 데이터도 변경될 수 있다는 의미이므로, 저장 전에 데이터 검증을 수행하도록 했다





6. 데이터 수집 및 자동 갱신

6.1. 최신 회차 확인

def get_latest_round():
    results = request_numbers()
 
    if results is None or len(results) != 1:
        print("최신 회차를 확인할 수 없습니다.")
        return None
 
    return results[0]["round"]

API에 매개변수를 전달하지 않으면 최신 회차의 당첨 결과가 반환된다. 따라서 별도로 추첨 날짜를 계산할 필요 없이 현재 공개된 최신 회차를 확인할 수 있다

6.2. 저장된 회차 조회

def get_saved_rounds():
    with sqlite3.connect(DB_PATH) as conn:
        cursor = conn.execute(
            "SELECT round FROM lotto"
        )
 
        return {
            row[0]
            for row in cursor.fetchall()
        }

데이터베이스에 저장된 회차 번호를 집합으로 반환한다

집합을 사용하는 이유는 회차의 존재 여부를 확인하고, 누락된 회차를 계산하기 편리하기 때문이다

6.3. 전체 회차 수집

def collect_all_numbers():
    print("전체 당첨 번호를 조회합니다.")
 
    results = request_numbers("all")
 
    if results is None or not results:
        print("전체 데이터 수집에 실패했습니다.")
        return False
 
    rounds = [
        result["round"]
        for result in results
    ]
 
    if len(rounds) != len(set(rounds)):
        print("API 응답에 중복된 회차가 있습니다.")
        return False
 
    latest = max(rounds)
 
    missing = (
        set(range(1, latest + 1))
        - set(rounds)
    )
 
    if missing:
        print(f"누락된 회차: {sorted(missing)}")
        return False
 
    saved_count = save_numbers(results)
 
    print(f"전체 {len(results)}개 회차 확인")
    print(f"데이터베이스 처리 행 수: {saved_count}")
 
    return True

전체 회차 수집 기능은 API에서 데이터를 가져온 뒤 다음 과정을 수행한다

  1. API 응답에서 당첨 번호 데이터 추출
  2. 각 회차의 데이터 유효성 검사
  3. 중복된 회차가 있는지 확인
  4. 1회부터 최신 회차까지 누락된 회차가 있는지 확인
  5. 검증을 통과한 데이터를 SQLite에 저장

API에서 여러 회차를 한 번에 받아오더라도 일부 데이터가 누락되거나 중복될 가능성을 고려해 별도의 검증 단계를 추가했다

6.4. 누락된 회차 자동 수집

def update_numbers():
    latest_round = get_latest_round()
 
    if latest_round is None:
        return
 
    saved_rounds = get_saved_rounds()
 
    missing = (
        set(range(1, latest_round + 1))
        - saved_rounds
    )
 
    if not missing:
        print(
            f"이미 최신 상태입니다. "
            f"({latest_round}회)"
        )
        return
 
    print(f"누락된 회차: {len(missing)}개")
 
    if len(missing) > 10:
        collect_all_numbers()
        return
 
    for round_number in sorted(missing):
        results = request_numbers(round_number)
 
        if results is None:
            print(f"{round_number}회 수집 실패")
            return
 
        matched = [
            result for result in results
            if result["round"] == round_number
        ]
 
        if not matched:
            print(f"{round_number}회 데이터가 없습니다.")
            return
 
        save_numbers(matched)
        print(f"{round_number}회 저장 완료")

update_numbers()는 데이터베이스에 저장된 회차와 API에서 확인한 최신 회차를 비교한다.

누락된 회차가 없다면 별도의 수집 작업을 수행하지 않는다. 반대로 누락된 회차가 10개를 초과하면 전체 데이터를 일괄 조회하고, 10개 이하라면 해당 회차만 개별적으로 조회한다

이렇게 구현하면 최초 실행 시에는 전체 데이터를 효율적으로 수집하고, 이후에는 새로운 회차만 추가할 수 있다





7. 실행 및 결과 확인

7.1 메인 함수

def main():
    create_database()
    update_numbers()
    show_database_status()
 
 
if __name__ == "__main__":
    main()

프로그램은 다음 순서로 실행된다

  1. 데이터베이스 및 테이블 생성
  2. 최신 회차 확인
  3. 누락된 회차 수집
  4. 데이터베이스 상태 출력

7.2. 데이터베이스 상태 출력

def show_database_status():
    saved = get_saved_rounds()
 
    if not saved:
        print("저장된 데이터가 없습니다.")
        return
 
    latest = max(saved)
 
    missing = (
        set(range(1, latest + 1))
        - saved
    )
 
    print("\n===== 데이터베이스 현황 =====")
    print(f"마지막 저장 회차: {latest}회")
    print(f"저장된 회차: {len(saved)}개")
    print(f"누락된 회차: {len(missing)}개")
 
    if missing:
        print(f"누락 목록: {sorted(missing)}")

7.3. 실제 실행 결과

기존 데이터베이스를 비운 상태에서 프로그램을 실행해 전체 회차 수집 및 저장 기능을 검증했다. 실제 콘솔 출력은 다음과 같다

누락된 회차: 1246개
전체 당첨 번호를 조회합니다.
전체 1246개 회차 확인
데이터베이스 처리 행 수: 1246
 
===== 데이터베이스 현황 =====
마지막 저장 회차: 1246회
저장된 회차: 1246개
누락된 회차: 0개

실행 결과

항목결과
최신 회차1246회
수집 대상1~1246회
API에서 확인한 데이터1,246개
데이터베이스 처리 행 수1,246개
실제 저장된 회차1,246개
누락된 회차0개

프로그램은 먼저 데이터베이스에 저장되지 않은 1,246개 회차를 확인했다. 이후 전체 회차 조회 API를 이용해 데이터를 수집하고, 각 회차의 데이터를 검증한 뒤 SQLite에 저장했다

최종적으로 데이터베이스에 1회부터 1246회까지 총 1,246개 회차가 저장되었으며, 누락된 회차가 없음을 확인했다

이를 통해 전체 당첨 번호 수집부터 데이터베이스 저장 및 누락 검사까지 정상적으로 동작하는 것을 검증했다





8. 마무리

이번 개발에서는 LottoLab의 첫 번째 기능인 로또 당첨 번호 수집 및 데이터베이스 관리 시스템을 구현했다

Python의 requests를 이용해 동행복권 웹사이트에서 당첨 번호를 가져오고, 수집한 데이터를 검증한 뒤 SQLite에 저장하는 구조를 만들었다. 또한 API의 전체 회차 조회 기능을 활용해 최초 데이터 수집에 필요한 요청 횟수를 줄이고, 이후에는 저장된 회차와 최신 회차를 비교해 필요한 데이터만 갱신하도록 구현했다

실제 실행을 통해 역대 1,246개 회차를 SQLite에 저장하고 누락된 회차가 없음을 확인했다

현재는 회차 번호의 중복과 누락, 당첨 번호의 기본적인 유효성을 검사하지만, 추첨 날짜의 실제 유효성이나 서로 다른 데이터 출처 간의 일치 여부까지 확인하는 기능은 아직 구현하지 않았다. 이 부분은 이후 데이터 신뢰성을 높이는 과정에서 추가로 개선할 수 있을 것이다

다음 개발 단계에서는 수집한 데이터를 바탕으로 로또 번호별 출현 빈도와 기초 통계 분석 기능을 구현할 예정이다

단순히 어떤 번호가 많이 등장했는지 확인하는 것에서 나아가, 실제 출현 횟수가 확률적으로 기대되는 값과 얼마나 차이가 나는지도 분석해 볼 생각이다

그래프 뷰