همگام‌سازی روزانهٔ دادهٔ بورس با پایگاه دادهٔ خودتان

برای همگام‌سازی روزانهٔ دادهٔ بورس با پایگاه دادهٔ خودتان سه کار لازم است: یک بار کل سابقه را با صفحه‌بندی تاریخی بگیرید، هر روز پس از ساعت ۱۸:۱۵ تهران فقط روزهای تازه را بگیرید، و هر ردیف را با upsert روی کلید (ins_code, d) بنویسید. با این الگو، اجرای دوباره یا نیمه‌کارهٔ برنامه هیچ ردیفی را تکراری یا خراب نمی‌کند.

این نوشته همین الگو را با پایتون و SQLite در یک فایل پیاده می‌کند. همان منطق در PostgreSQL یا هر پایگاه دادهٔ دیگری کار می‌کند.

چهار قاعدهٔ این الگو

1. پرکردن سابقه از جدید به قدیم. هر درخواست حداکثر ۵۰۰۰ ردیف می‌دهد و limit جدیدترین ردیف‌های بازه را نگه می‌دارد. پس صفحهٔ اول، ۵۰۰۰ روز آخر است. برای صفحهٔ بعد، end را روز پیش از قدیمی‌ترین ردیف دریافتی بگذارید. وقتی صفحه‌ای کمتر از ۵۰۰۰ ردیف داشت، سابقه تمام شده است.
2. به‌روزرسانی روزانه پس از ۱۸:۱۵. دادهٔ پایان روز، ساعت ۱۸:۱۵ روزهای شنبه تا چهارشنبه به وقت تهران به‌روز می‌شود. برنامهٔ خود را کمی بعد از آن اجرا کنید.
3. چند روز آخر را دوباره بگیرید. TSETMC بعضی ارقام پایان روز را دیر منتشر می‌کند و دادهٔ آخرین روز ممکن است تا اجرای بعدی تغییر کند. هر بار از چند روز پیش از آخرین ردیف خودتان شروع کنید.
4. upsert روی (ins_code, d). هر نماد در هر روز یک ردیف دارد. اگر ردیف هست، به‌روز می‌شود؛ اگر نیست، ساخته می‌شود.

قیمت خام را ذخیره کنم یا تعدیل‌شده؟

قیمت خام را ذخیره کنید و رویدادهای تعدیل را در جدول جدا نگه دارید. قیمت خام یک روز دیگر تغییر نمی‌کند، اما سری تعدیل‌شده با هر افزایش سرمایه یا سود نقدی تازه از اول عوض می‌شود. اگر تعدیل‌شده را ذخیره کنید، باید پس از هر رویداد کل سابقهٔ آن نماد را دوباره بگیرید.

با جدول رویدادها، تعدیل را هنگام خواندن می‌سازید. تابع آن در قیمت تعدیل‌شده چیست؟ آمده است.

کد کامل

این فایل را با نام sync.py ذخیره کنید. فقط به requests نیاز دارد.

import os
import sqlite3
import time
from datetime import date, timedelta

import requests

BASE = "https://poolick.ir"
HEADERS = {"Authorization": f"Bearer {os.environ['ROSHA_KEY']}"}
PAGE = 5000  # the API's maximum rows per request
OVERLAP_DAYS = 7  # re-pull the last few days on every run
INS_CODES = ["46348559193224090"]  # فولاد; keep ins_code as text

COLS = ["d", "open", "high", "low", "close", "last", "prev_close",
        "volume", "value", "trades", "jdate"]

db = sqlite3.connect("market.db")
db.executescript("""
CREATE TABLE IF NOT EXISTS daily (
    ins_code TEXT NOT NULL, d TEXT NOT NULL,
    open REAL, high REAL, low REAL, close REAL, last REAL, prev_close REAL,
    volume REAL, value REAL, trades INTEGER, jdate TEXT,
    PRIMARY KEY (ins_code, d)
);
CREATE TABLE IF NOT EXISTS adjustment (
    ins_code TEXT NOT NULL, d TEXT NOT NULL,
    close_adjusted REAL, close_unadjusted REAL,
    PRIMARY KEY (ins_code, d)
);
""")

UPSERT = (
    f"INSERT INTO daily (ins_code, {', '.join(COLS)}) "
    f"VALUES (?, {', '.join('?' for _ in COLS)}) "
    "ON CONFLICT (ins_code, d) DO UPDATE SET "
    + ", ".join(f"{c} = excluded.{c}" for c in COLS[1:])
)


def get(path, **kwargs):
    """GET with one retry on 429, waiting as long as Retry-After says."""
    for _ in range(2):
        r = requests.get(BASE + path, headers=HEADERS, timeout=30, **kwargs)
        if r.status_code != 429:
            break
        time.sleep(int(r.headers.get("Retry-After", "1")))
    r.raise_for_status()
    return r.json()


def upsert(ins_code, rows):
    with db:  # one transaction per page
        db.executemany(UPSERT, [(ins_code, *(r[c] for c in COLS)) for r in rows])
    return len(rows)


def backfill(ins_code):
    """Whole history, newest page first, paging back by date."""
    total, end = 0, None
    while True:
        params = {"limit": PAGE}
        if end:
            params["end"] = end
        rows = get(f"/v1/instruments/{ins_code}/daily", params=params)["rows"]
        total += upsert(ins_code, rows)
        if len(rows) < PAGE:
            return total
        # rows are oldest first: the next page ends the day before rows[0]
        end = (date.fromisoformat(rows[0]["d"]) - timedelta(days=1)).isoformat()


def incremental(ins_code):
    """Rows since the last stored day, minus a few days of overlap."""
    last = db.execute(
        "SELECT max(d) FROM daily WHERE ins_code = ?", (ins_code,)
    ).fetchone()[0]
    if last is None:
        return backfill(ins_code)
    start = (date.fromisoformat(last) - timedelta(days=OVERLAP_DAYS)).isoformat()
    rows = get(
        f"/v1/instruments/{ins_code}/daily", params={"start": start, "limit": PAGE}
    )["rows"]
    return upsert(ins_code, rows)


def sync_adjustments(ins_code):
    """The event list is small: replace it as a whole."""
    events = get(f"/v1/instruments/{ins_code}/adjustments")
    with db:
        db.execute("DELETE FROM adjustment WHERE ins_code = ?", (ins_code,))
        db.executemany(
            "INSERT INTO adjustment VALUES (?, ?, ?, ?)",
            [(ins_code, e["d"], e["close_adjusted"], e["close_unadjusted"]) for e in events],
        )
    return len(events)


if __name__ == "__main__":
    for code in INS_CODES:
        written = incremental(code)
        events = sync_adjustments(code)
        stored, first, last = db.execute(
            "SELECT count(*), min(d), max(d) FROM daily WHERE ins_code = ?", (code,)
        ).fetchone()
        print(f"{code}: wrote {written}, stored {stored} rows ({first} to {last}), {events} events")

اجرای اول کل سابقه را می‌گیرد. اجرای دوم فقط چند روز آخر را دوباره می‌نویسد. خروجی واقعی دو اجرای پشت سر هم در ۲۳ آبان ۱۴۰۵:

46348559193224090: wrote 4680, stored 4680 rows (2007-03-11 to 2026-11-11), 23 events
46348559193224090: wrote 6, stored 4680 rows (2007-03-11 to 2026-11-11), 23 events

تعداد ردیف‌های ذخیره‌شده در اجرای دوم تغییر نکرده است. این همان تکرارپذیری است.

چرا این کد در اجرای دوباره خراب نمی‌شود؟

• کلید اصلی (ins_code, d). پایگاه داده اجازهٔ ردیف تکراری نمی‌دهد و ON CONFLICT ... DO UPDATE ردیف موجود را با مقدار تازه جایگزین می‌کند.
• هر صفحه یک تراکنش. اگر برنامه وسط کار قطع شود، صفحه‌های نوشته‌شده سالم می‌مانند.
• نقطهٔ شروع از خود جدول می‌آید. برنامه جای خود را در فایل جدا نگه نمی‌دارد. max(d) همیشه راست می‌گوید.
• رویدادها یک‌جا جایگزین می‌شوند. حذف و درج در یک تراکنش است و جدول هیچ‌وقت نیمه‌خالی دیده نمی‌شود.

یک نکته دربارهٔ قطع‌شدن وسط پرکردن سابقه: چون صفحه‌ها از جدید به قدیم نوشته می‌شوند، max(d) پس از اولین صفحه پر است و اجرای بعدی فقط به‌روزرسانی روزانه می‌کند. نمادهایی که بیش از ۵۰۰۰ روز سابقه دارند کم هستند. برای آن‌ها backfill را تا پایان اجرا کنید یا تعداد ردیف‌ها را با first_trade_date نماد مقایسه کنید.

زمان‌بندی اجرا

دادهٔ پایان روز ساعت ۱۸:۱۵ تهران به‌روز می‌شود. اجرای ۱۸:۴۵ حاشیهٔ کافی دارد. با cron:

CRON_TZ=Asia/Tehran
45 18 * * 6,0-3 cd /opt/market && . ./rosha.env && python3 sync.py >> sync.log 2>&1

6,0-3 یعنی شنبه تا چهارشنبه. فایل rosha.env خط export ROSHA_KEY=... را دارد و نباید در مخزن کد باشد. اگر cron شما CRON_TZ را نمی‌شناسد، ساعت را به وقت سرور تبدیل کنید.

بازار در تعطیلات رسمی بسته است. اجرا در روز تعطیل ضرری ندارد: برنامه همان چند روز آخر را دوباره می‌نویسد. اگر می‌خواهید در روز تعطیل اصلاً درخواست نزنید، تقویم معاملاتی را بپرسید.

چند درخواست مصرف می‌شود؟

این الگو برای هر نماد در هر روز دو درخواست می‌فرستد: یکی برای قیمت و یکی برای رویدادها. با سقف پلن رایگان (۱٬۰۰۰ درخواست در روز تا آبان ۱۴۰۵) حدود ۵۰۰ نماد را می‌توانید هر روز همگام کنید. برای کم‌کردن مصرف، رویدادها را هفته‌ای یک بار بگیرید.

تابع get در پاسخ 429 به اندازهٔ Retry-After صبر می‌کند. اگر نمادهای زیادی دارید، میان درخواست‌ها مکث کوتاهی بگذارید تا به سقف دقیقه‌ای نرسید. سقف هر پلن در محدودیت‌ها آمده است.

گسترش به داده‌های دیگر

همین الگو برای بقیهٔ نقطه‌های پایانی تاریخی کار می‌کند. فقط کلید جدول فرق دارد:

سابقهٔ صندوق‌ها یک تفاوت دارد. فیپیران مقادیر گذشته را بازنگری می‌کند و روشا هر جمعه ساعت ۰۲:۰۰ کل سابقهٔ صندوق‌ها را از نو می‌سازد. برای صندوق‌ها بازهٔ هم‌پوشانی را بلندتر بگیرید یا هفته‌ای یک بار کل سابقه را دوباره بکشید.

راهنمای مرجع این الگو در همگام‌سازی روزانه و زمان به‌روزرسانی هر داده در تازگی داده است.