การเก็บถาวรข้อมูลประวัติคริปโตเคอร์เรนซี: วิธีจัดเก็บข้อมูลจาก API ของตลาดแลกเปลี่ยนอย่างถาวร

หลายคนที่เริ่มสนใจการลงทุนคริปโตเคอร์เรนซีมักอยากเก็บข้อมูลราคาย้อนหลังไว้วิเคราะห์ แต่เจอปัญหาว่าข้อมูลเก่าๆ หายไปจากเว็บไซต์ วันนี้ผมจะสอนวิธีดึงข้อมูลจากตลาดแลกเปลี่ยนมาเก็บไว้ใช้งานเองแบบทีละขั้นตอน ไม่ต้องมีความรู้เรื่องโค้ดมาก่อนก็ทำได้

ทำความรู้จักกับ API คืออะไร

นึกภาพง่ายๆ API ก็เหมือนปลั๊กอินที่เชื่อมต่อระหว่างโปรแกรมสองตัว ข้อมูลราคาคริปโตที่คุณเห็นบนเว็บไซต์ ล้วนมาจาก API ทั้งนั้น ตลาดแลกเปลี่ยนอย่าง Binance, Coinbase หรือ Kraken ล้วนมี API ให้ใช้ฟรี คุณสามารถขอข้อมูลราคา ปริมาณการซื้อขาย และข้อมูลอื่นๆ มาใช้งานได้

เตรียมอะไรบ้างก่อนเริ่มต้น

• คอมพิวเตอร์ที่ติดตั้ง Python 3.8 ขึ้นไป (ดาวน์โหลดได้ฟรีจาก python.org)
• บัญชีตลาดแลกเปลี่ยนที่รองรับ API (แนะนำ Binance รองรับภาษาไทย)
• ความตั้งใจเรียนรู้สัก 30 นาที

ขั้นตอนที่ 1: สมัครบัญชีและขอ API Key

ไปที่เว็บไซต์ Binance สมัครสมาชิกแล้วยืนยันตัวตน จากนั้นไปที่หน้า API Management สร้าง Key ใหม่ ตั้งชื่ออะไรก็ได้ เช่น "crypto-data-collector" ระบบจะให้ API Key และ Secret Key มาคู่หนึ่ง คัดลอกเก็บไว้ในที่ปลอดภัย อย่าแชร์ให้ใครเด็ดขาด

หมายเหตุ: หากเป็นบัญชีที่มีเงินจริง แนะนำตั้งค่า IP whitelist เพื่อความปลอดภัย

ขั้นตอนที่ 2: ติดตั้งโปรแกรมที่จำเป็น

เปิด Command Prompt หรือ Terminal แล้วพิมพ์คำสั่งนี้

pip install python-binance pandas sqlalchemy sqlite3 schedule requests

รอให้ติดตั้งเสร็จ ปกติใช้เวลาไม่เกิน 2 นาที โปรแกรมเหล่านี้คือเครื่องมือที่ช่วยดึงข้อมูล จัดการตาราง และเก็บข้อมูลลงฐานข้อมูล

ขั้นตอนที่ 3: เขียนโค้ดพื้นฐานดึงข้อมูลราคา

สร้างไฟล์ใหม่ชื่อ crypto_collector.py แล้วคัดลอกโค้ดด้านล่างไปวาง

from binance.client import Client
import pandas as pd
from datetime import datetime
import time

ใส่ API Key ของคุณที่นี่
API_KEY = "YOUR_BINANCE_API_KEY"
SECRET_KEY = "YOUR_BINANCE_SECRET_KEY"

เชื่อมต่อกับ Binance
client = Client(API_KEY, SECRET_KEY)

def get_current_price(symbol="BTCUSDT"):
    """ดึงราคาปัจจุบันของเหรียญ"""
    ticker = client.get_symbol_ticker(symbol=symbol)
    price = float(ticker['price'])
    return {
        'symbol': symbol,
        'price': price,
        'timestamp': datetime.now()
    }

ทดสอบดึงราคา Bitcoin
result = get_current_price("BTCUSDT")
print(f"ราคา BTC/USDT ปัจจุบัน: ${result['price']:,.2f}")
print(f"เวลา: {result['timestamp']}")

รันโค้ดด้วยคำสั่ง python crypto_collector.py หากขึ้นราคามาแสดงว่าเชื่อมต่อสำเร็จ

ขั้นตอนที่ 4: ดึงข้อมูลประวัติราคาย้อนหลัง

ต่อไปจะดึงข้อมูลราคาย้อนหลังหลายวันมาเก็บไว้ ซึ่งเป็นหัวใจสำคัญของการวิเคราะห์

def get_historical_klines(symbol="BTCUSDT", interval="1h", days=30):
    """ดึงข้อมูลราคาย้อนหลัง"""
    # คำนวณจำนวนเทียบของ interval
    interval_map = {
        "1m": Client.KLINE_INTERVAL_1MINUTE,
        "5m": Client.KLINE_INTERVAL_5MINUTE,
        "15m": Client.KLINE_INTERVAL_15MINUTE,
        "1h": Client.KLINE_INTERVAL_1HOUR,
        "4h": Client.KLINE_INTERVAL_4HOUR,
        "1d": Client.KLINE_INTERVAL_1DAY
    }
    
    # ดึงข้อมูลจาก Binance
    klines = client.get_historical_klines(
        symbol, 
        interval_map[interval], 
        f"{days} days ago UTC"
    )
    
    # แปลงเป็น DataFrame
    df = pd.DataFrame(klines, columns=[
        'open_time', 'open', 'high', 'low', 'close', 'volume',
        'close_time', 'quote_volume', 'trades', 'taker_buy_base',
        'taker_buy_quote', 'ignore'
    ])
    
    # เลือกเฉพาะคอลัมน์ที่ต้องการ
    df = df[['open_time', 'open', 'high', 'low', 'close', 'volume']]
    
    # แปลง timestamp เป็นวันที่
    df['date'] = pd.to_datetime(df['open_time'], unit='ms')
    
    # แปลงข้อมูลเป็นตัวเลข
    for col in ['open', 'high', 'low', 'close', 'volume']:
        df[col] = pd.to_numeric(df[col])
    
    return df

ดึงข้อมูล 30 วัน ราคา Bitcoin
btc_data = get_historical_klines("BTCUSDT", "1h", 30)
print(f"ได้ข้อมูลทั้งหมด {len(btc_data)} แถว")
print(btc_data.tail())

ขั้นตอนที่ 5: บันทึกข้อมูลลงฐานข้อมูล SQLite

เก็บข้อมูลไว้ในไฟล์ฐานข้อมูล SQLite ซึ่งเป็นไฟล์เดียวเก็บได้หลายล้านแถว

import sqlite3
from sqlalchemy import create_engine

def save_to_database(data, symbol="BTCUSDT", table_name="crypto_prices"):
    """บันทึกข้อมูลลง SQLite"""
    db_path = "crypto_data.db"
    
    # สร้าง database connection
    engine = create_engine(f'sqlite:///{db_path}')
    
    # เพิ่ม symbol column
    data['symbol'] = symbol
    
    # บันทึกลงตาราง (append ต่อจากข้อมูลเดิม)
    data.to_sql(table_name, engine, if_exists='append', index=False)
    
    print(f"บันทึก {len(data)} แถว ลง {table_name} เรียบร้อย")

ดึงและบันทึกข้อมูล
btc_data = get_historical_klines("BTCUSDT", "1h", 7)
save_to_database(btc_data, "BTCUSDT")

ดึงข้อมูล Ethereum ด้วย
eth_data = get_historical_klines("ETHUSDT", "1h", 7)
save_to_database(eth_data, "ETHUSDT")

print("เสร็จสิ้นการเก็บข้อมูลคริปโต")

ขั้นตอนที่ 6: ตั้งเวลาเก็บข้อมูลอัตโนมัติ

แทนที่จะมานั่งรันโค้ดเองทุกครั้ง ตั้งให้ระบบเก็บข้อมูลอัตโนมัติทุกชั่วโมง

import schedule
import time

def job():
    """เก็บข้อมูลราคาทุกครั้งที่รัน"""
    print(f"เริ่มเก็บข้อมูล {datetime.now()}")
    
    coins = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT"]
    
    for coin in coins:
        try:
            data = get_historical_klines(coin, "1h", 1)
            save_to_database(data, coin)
            print(f"  ✓ {coin} เก็บสำเร็จ")
            time.sleep(2)  # รอ 2 วินาทีระหว่างแต่ละเหรียญ
        except Exception as e:
            print(f"  ✗ {coin} ผิดพลาด: {e}")
    
    print(f"เสร็จสิ้น {datetime.now()}\n")

ตั้งเวลาเก็บทุก 1 ชั่วโมง
schedule.every(1).hours.do(job)

ทดสอบรันทันที 1 ครั้ง
job()

วนรอบต่อไป
print("กำลังทำงานอัตโนมัติ กด Ctrl+C เพื่อหยุด")
while True:
    schedule.run_pending()
    time.sleep(60)

วิธีดูข้อมูลที่เก็บไว้

เมื่อต้องการนำข้อมูลมาใช้วิเคราะห์ ใช้โค้ดนี้

import pandas as pd
from sqlalchemy import create_engine

def load_data(symbol="BTCUSDT", days=7):
    """โหลดข้อมูลจากฐานข้อมูล"""
    engine = create_engine('sqlite:///crypto_data.db')
    
    query = f"""
    SELECT date, open, high, low, close, volume
    FROM crypto_prices
    WHERE symbol = '{symbol}'
    ORDER BY date DESC
    LIMIT {days * 24}
    """
    
    df = pd.read_sql(query, engine)
    df['date'] = pd.to_datetime(df['date'])
    return df

โหลดข้อมูล Bitcoin 7 วันล่าสุด
btc = load_data("BTCUSDT", 7)

print("ราคา Bitcoin 7 วันล่าสุด:")
print(btc.to_string())

หาค่าเฉลี่ย
print(f"\nราคาเฉลี่ย: ${btc['close'].mean():,.2f}")
print(f"ราคาสูงสุด: ${btc['high'].max():,.2f}")
print(f"ราคาต่ำสุด: ${btc['low'].min():,.2f}")

รวมโค้ดทั้งหมดเป็นไฟล์เดียว

นี่คือโค้ดสมบูรณ์ที่รวมทุกอย่างเข้าด้วยกัน บันทึกเป็น full_collector.py

from binance.client import Client
import pandas as pd
from sqlalchemy import create_engine
import schedule
import time
from datetime import datetime

========== ตั้งค่า ==========
API_KEY = "YOUR_BINANCE_API_KEY"
SECRET_KEY = "YOUR_BINANCE_SECRET_KEY"
COINS = ["BTCUSDT", "ETHUSDT", "BNBUSDT"]
INTERVAL = "1h"
DB_PATH = "crypto_archive.db"
==============================

client = Client(API_KEY, SECRET_KEY)

def get_historical_klines(symbol, days=1):
    klines = client.get_historical_klines(
        symbol,
        INTERVAL,
        f"{days} day ago UTC"
    )
    df = pd.DataFrame(klines, columns=[
        'open_time', 'open', 'high', 'low', 'close', 'volume',
        'close_time', 'quote_volume', 'trades'
    ])
    df = df[['open_time', 'open', 'high', 'low', 'close', 'volume']]
    df['date'] = pd.to_datetime(df['open_time'], unit='ms')
    df['symbol'] = symbol
    for col in ['open', 'high', 'low', 'close', 'volume']:
        df[col] = pd.to_numeric(df[col])
    return df[['date', 'symbol', 'open', 'high', 'low', 'close', 'volume']]

def save_to_database(data):
    engine = create_engine(f'sqlite:///{DB_PATH}')
    data.to_sql('prices', engine, if_exists='append', index=False)

def collect_all():
    print(f"[{datetime.now()}] เริ่มเก็บข้อมูล...")
    for coin in COINS:
        try:
            data = get_historical_klines(coin, 1)
            save_to_database(data)
            print(f"  ✓ {coin}: {len(data)} rows")
            time.sleep(1)
        except Exception as e:
            print(f"  ✗ {coin}: {e}")
    print(f"[{datetime.now()}] เสร็จสิ้น\n")

ทดสอบทันที 1 ครั้ง
collect_all()

ตั้งเวลาทุก 1 ชั่วโมง
schedule.every(1).hours.do(collect_all)

while True:
    schedule.run_pending()
    time.sleep(60)

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

กรณีที่ 1: API Key ไม่ถูกต้อง หรือถูกระงับ

อาการ: ได้รับข้อความ APIError: -1021: Timestamp for this request was not received

สาเหตุ: เวลาบนเครื่องคอมพิวเตอร์ไม่ตรงกับเซิร์ฟเวอร์ หรือ API Key หมดอายุ

# วิธีแก้ไข: ตรวจสอบและแก้เวลาเครื่อง
Windows: ตั้งค่า > วันที่และเวลา > ตั้งเวลาอัตโนมัติ

หรือใช้โค้ดแก้ timestamp
from binance.client import Client

ใส่ค่า time_offset ปรับเวลาให้ตรง
client = Client(API_KEY, SECRET_KEY, time_offset=0)

หากยังไม่ได้ ลองสร้าง API Key ใหม่จากเว็บ Binance

กรณีที่ 2: ถูกจำกัดการใช้งาน (Rate Limit)

อาการ: ได้รับข้อความ APIError: -1003: Too many requests

สาเหตุ: ส่งคำขอมากเกินไปในเวลาสั้น

import time

def safe_get_data(symbol, retries=3):
    """ดึงข้อมูลอย่างปลอดภัยพร้อมรอเมื่อถูกจำกัด"""
    for i in range(retries):
        try:
            return get_historical_klines(symbol, 1)
        except Exception as e:
            if "Too many requests" in str(e):
                wait_time = (i + 1) * 10  # รอ 10, 20, 30 วินาที
                print(f"รอ {wait_time} วินาทีก่อนลองใหม่...")
                time.sleep(wait_time)
            else:
                raise e
    return None

ใช้แทน get_historical_klines โดยตรง

กรณีที่ 3: ข้อมูลซ้ำเมื่อรันหลายครั้ง

อาการ: ฐานข้อมูลมีข้อมูลซ้ำกันหลายแถว

สาเหตุ: ใช้ if_exists='append' ต่อท้ายโดยไม่มีการตรวจสอบ

def save_to_database_no_duplicate(data):
    """บันทึกโดยไม่ให้ข้อมูลซ้ำ"""
    engine = create_engine(f'sqlite:///{DB_PATH}')
    
    # โหลดข้อมูลเดิมมาตรวจสอบ
    existing = pd.read_sql('prices', engine)
    
    if len(existing) > 0:
        # กรองเอาเฉพาะข้อมูลใหม่
        new_data = data[~data['date'].isin(existing['date'])]
        if len(new_data) > 0:
            new_data.to_sql('prices', engine, if_exists='append', index=False)
            print(f"เพิ่ม {len(new_data)} แถวใหม่")
        else:
            print("ไม่มีข้อมูลใหม่")
    else:
        data.to_sql('prices', engine, if_exists='replace', index=False)
        print("สร้างตารางใหม่")

กรณีที่ 4: เหรียญที่ระบุไม่มีอยู่จริง

อาการ: ได้รับข้อผิดพลาด APIError: -1121: Symbol not found

สาเหตุ: ชื่อเหรียญไม่ตรงตามรูปแบบ หรือเหรียญนั้นถูกถอดออกจากตลาด

def get_valid_symbols():
    """ดึงรายชื่อเหรียญที่มีอยู่จริงทั้งหมด"""
    exchange_info = client.get_exchange_info()
    symbols = [s['symbol'] for s in exchange_info['symbols']]
    return symbols

ดึงรายชื่อเหรียญทั้งหมด
valid_symbols = get_valid_symbols()
print(f"มีเหรียญทั้งหมด {len(valid_symbols)} ชนิด")

ตรวจสอบก่อนใช้งาน
test_coin = "DOGEUSDT"
if test_coin in valid_symbols:
    print(f"{test_coin} มีอยู่จริง พร้อมใช้งาน")
else:
    print(f"{test_coin} ไม่มีอยู่ในตลาด")

เหมาะกับใคร / ไม่เหมาะกับใคร

เหมาะกับ	ไม่เหมาะกับ
นักลงทุนที่ต้องการวิเคราะห์ราคาย้อนหลังด้วยตัวเอง	คนที่ต้องการดูราคาแบบเรียลไทม์ตลอดเวลา
นักพัฒนาที่กำลังสร้างเครื่องมือวิเคราะห์คริปโต	คนที่ไม่มีคอมพิวเตอร์ใช้งานเป็นประจำ
เทรดเดอร์ที่ใช้กลยุทธ์ Backtesting	คนที่ไม่มีเวลาดูแลระบบแม้จะเป็นอัตโนมัติ
นักวิจัยหรือนักศึกษาที่ศึกษาเรื่องการเงิน	คนที่ต้องการดูข้อมูลเพียงครั้งคราว

ราคาและ ROI

รายการ	ราคา	หมายเหตุ
Binance API	ฟรี	เพียงพอสำหรับผู้เริ่มต้น
Python และไลบรารี	ฟรี	โอเพนซอร์สทั้งหมด
ค่าไฟฟ้าคอมพิวเตอร์	ประมาณ 50-100 บาท/เดือน	หากเปิดรันตลอดเวลา
พื้นที่จัดเก็บข้อมูล 1 ปี	น้อยกว่า 500 MB	ข้อมูล 4 เหรียญ ทุก 1 ชั่วโมง

ทำไมต้องเลือก HolySheep

หากคุณวางแผนนำข้อมูลที่เก็บได้ไปวิเคราะห์ด้วย AI เพื่อหาแนวโน้มราคา หรือสร้างรายงานอัตโนมัติ HolySheep AI คือตัวเลือกที่คุ้มค่าที่สุดในตลาดปี 2026

บริการ	ราคาเดิม	HolySheep	ประหยัด
GPT-4.1	$8/MTok	$8/MTok	อัตราแลกเปลี่ยนพิเศษ ¥1=$1
Claude Sonnet 4.5	$15/MTok	$15/MTok	รองรับ WeChat/Alipay
Gemini 2.5 Flash	$2.50/MTok	$2.50/MTok	เครดิตฟรีเมื่อลงทะเบียน
DeepSeek V3.2	$0.42/MTok	$0.42/MTok	ความหน่วงต่ำกว่า 50ms

นอกจากราคาที่ประหยัดกว่าถึง 85% เมื่อเทียบกับการซื้อจากแพลตฟอร์มอื่นในสกุลเงินบาทแล้ว ระบบยังรองรับการชำระเงินผ่าน WeChat Pay และ Alipay ซึ่งสะดวกมากสำหรับคนไทยที่มีเพื่อนหรือญาติในจีน �