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

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

طرح کلی چیست؟

گام	چه زمانی	درخواست
بارگیری اولیه	یک بار برای هر نماد	/daily?limit=5000، سپس صفحه‌های قدیمی‌تر با end
همگام‌سازی روزانه	شنبه تا چهارشنبه، پس از ۱۸:۱۵ تهران	/daily?start=<آخرین روز ذخیره‌شده منهای چند روز>
نوشتن	هر بار	upsert روی (ins_code, d)

سه تصمیم که کد زیر بر آن‌ها بنا شده است:

• قیمت خام ذخیره کنید. قیمت تعدیل‌شدهٔ روزهای گذشته با هر رویداد تعدیل تازه عوض می‌شود، ولی قیمت خام ثابت می‌ماند. قیمت تعدیل‌شده را هنگام نیاز بگیرید یا از روی رویدادها بسازید؛ بخش «قیمت تعدیل‌شده را چطور نگه دارم؟» را ببینید.
• ins_code را متن ذخیره کنید و نماد را با ins_code مشخص کنید، نه با نماد فارسی که ممکن است مبهم باشد.
• کار را تکرارپذیر (idempotent) بنویسید. اجرای دوباره نباید ردیف تکراری بسازد؛ upsert همین را تضمین می‌کند.

چرا چند روز آخر را دوباره بگیرم؟

چون دادهٔ آخرین روز ممکن است پس از ۱۸:۱۵ تغییر کند. TSETMC بعضی ارقام پایان روز را دیر نهایی می‌کند و روشا آن‌ها را در اجرای بعدی اصلاح می‌کند. کد زیر هر بار از ۷ روز پیش از آخرین روز ذخیره‌شده شروع می‌کند؛ upsert ردیف‌های موجود را با مقدار تازه جایگزین می‌کند. جدول زمان‌ها در به‌روزرسانی داده‌ها است.

کد کامل با SQLite

این برنامه هم بارگیری اولیه را انجام می‌دهد و هم همگام‌سازی روزانه را: اگر نمادی در پایگاه داده ردیف نداشته باشد کل سابقه را می‌گیرد، وگرنه فقط روزهای تازه را. فقط به requests نیاز دارد (pip install requests).

"""Keep a local SQLite copy of ROSHA daily prices in sync.

Run once to backfill, then every trading day after 18:15 Asia/Tehran.
"""
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']}"}
DB_PATH = "rosha.db"
INS_CODES = ["46348559193224090"]  # فولاد; ins_code is text, never a number
PAGE = 5000  # the REST cap on `limit`
OVERLAP_DAYS = 7  # re-pull the last days: the latest day can still change

SCHEMA = """
CREATE TABLE IF NOT EXISTS daily (
    ins_code   TEXT NOT NULL,
    d          TEXT NOT NULL,  -- Gregorian ISO date
    open       REAL,
    high       REAL,
    low        REAL,
    close      REAL,
    last       REAL,
    prev_close REAL,
    volume     REAL,
    value      REAL,
    trades     INTEGER,
    PRIMARY KEY (ins_code, d)
)
"""

UPSERT = """
INSERT INTO daily (ins_code, d, open, high, low, close, last, prev_close, volume, value, trades)
VALUES (:ins_code, :d, :open, :high, :low, :close, :last, :prev_close, :volume, :value, :trades)
ON CONFLICT (ins_code, d) DO UPDATE SET
    open = excluded.open, high = excluded.high, low = excluded.low,
    close = excluded.close, last = excluded.last, prev_close = excluded.prev_close,
    volume = excluded.volume, value = excluded.value, trades = excluded.trades
"""


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 fetch_daily(ins_code, start=None):
    """Unadjusted daily rows from `start` (or from the first day), oldest first."""
    rows, end = [], None
    while True:
        params = {"limit": PAGE}
        if start:
            params["start"] = start
        if end:
            params["end"] = end
        batch = get(f"/v1/instruments/{ins_code}/daily", params=params)["rows"]
        rows = batch + rows  # each page is older than the one before
        if len(batch) < PAGE:
            return rows
        oldest = date.fromisoformat(batch[0]["d"])
        end = (oldest - timedelta(days=1)).isoformat()


def sync(con, ins_code):
    """Backfill if the instrument is new, otherwise pull only the recent days."""
    last = con.execute(
        "SELECT max(d) FROM daily WHERE ins_code = ?", (ins_code,)
    ).fetchone()[0]
    start = None
    if last:
        start = (date.fromisoformat(last) - timedelta(days=OVERLAP_DAYS)).isoformat()
    rows = fetch_daily(ins_code, start)
    with con:  # one transaction per instrument
        con.executemany(UPSERT, [{**r, "ins_code": ins_code} for r in rows])
    return len(rows)


def main():
    con = sqlite3.connect(DB_PATH)
    con.execute(SCHEMA)
    for ins_code in INS_CODES:
        n = sync(con, ins_code)
        total = con.execute(
            "SELECT count(*), max(d) FROM daily WHERE ins_code = ?", (ins_code,)
        ).fetchone()
        print(f"{ins_code}: upserted {n} rows, table has {total[0]} rows up to {total[1]}")
    con.close()


if __name__ == "__main__":
    main()

اجرای اول برای فولاد همهٔ ردیف‌ها را می‌نویسد. اجرای دوباره در همان روز فقط چند ردیف آخر را می‌گیرد و تعداد ردیف‌های جدول عوض نمی‌شود.

چه زمانی اجرا کنم؟

هر روز معاملاتی (شنبه تا چهارشنبه) پس از ساعت ۱۸:۱۵ به وقت تهران. دادهٔ روزانه در آن ساعت تازه می‌شود؛ اجرای زودتر، ردیف امروز را ندارد.

نمونهٔ crontab برای سروری که ساعت آن روی وقت تهران است. اگر ساعت سرور UTC است، ساعت را به ۱۵:۱۵ UTC تغییر دهید (تهران سه ساعت و نیم از UTC جلوتر است):

# minute hour day-of-month month day-of-week (6 = Saturday, 0-3 = Sunday to Wednesday)
45 18 * * 6,0,1,2,3  ROSHA_KEY="rosha_xxxxxxxx_your-secret" /usr/bin/python3 /opt/rosha/sync.py

• ساعت ۱۸:۴۵ نیم ساعت پس از زمان به‌روزرسانی است تا کار پایان روز تمام شده باشد.
• اجرا در روز تعطیل ضرری ندارد؛ فقط ردیف تازه‌ای نمی‌آید.
• اگر یک یا چند روز اجرا نشد، اجرای بعدی جاافتاده‌ها را می‌گیرد، چون از آخرین روز ذخیره‌شده شروع می‌کند.
• برای اینکه بدانید امروز روز معاملاتی بوده، /v1/calendar را ببینید.

با سهمیهٔ پلن چطور کنار بیایم؟

هر نماد در همگام‌سازی روزانه یک درخواست و در بارگیری اولیه معمولاً یک درخواست مصرف می‌کند (هر ۵٬۰۰۰ ردیف یک درخواست).

پلن (تا آبان ۱۴۰۵)	درخواست در روز	بیشترین تعداد نماد در همگام‌سازی روزانه
رایگان	1,000	حدود 1,000
پایه	10,000	حدود 10,000
حرفه‌ای	20,000	حدود 20,000

سقف دقیقه‌ای هم برقرار است: با پلن رایگان (۳۰ درخواست در دقیقه) تابع get پس از درخواست سی‌ام تا شروع دقیقهٔ بعد صبر می‌کند و ادامه می‌دهد. اگر سهمیهٔ روزانه تمام شود، پاسخ 429 quota_exceeded با Retry-After: 3600 است؛ get یک ساعت صبر می‌کند، دوباره 429 می‌گیرد و استثنا می‌دهد. برای فهرست بلند نمادها، بارگیری اولیه را بین چند روز پخش کنید. جزئیات در محدودیت‌ها و پلن‌ها.

برای PostgreSQL چه چیزی عوض می‌شود؟

فقط اتصال، نوع ستون‌ها و شکل جای‌نگهدارها. منطق fetch_daily و sync همان است و دستور ON CONFLICT در PostgreSQL هم همین شکل را دارد.

CREATE TABLE IF NOT EXISTS daily (
    ins_code   text    NOT NULL,
    d          date    NOT NULL,
    open       numeric,
    high       numeric,
    low        numeric,
    close      numeric,
    last       numeric,
    prev_close numeric,
    volume     numeric,
    value      numeric,
    trades     integer,
    PRIMARY KEY (ins_code, d)
);

با psycopg (نسخهٔ ۳؛ pip install "psycopg[binary]") این سه بخش برنامهٔ بالا را جایگزین کنید:

import psycopg

UPSERT = """
INSERT INTO daily (ins_code, d, open, high, low, close, last, prev_close, volume, value, trades)
VALUES (%(ins_code)s, %(d)s, %(open)s, %(high)s, %(low)s, %(close)s, %(last)s,
        %(prev_close)s, %(volume)s, %(value)s, %(trades)s)
ON CONFLICT (ins_code, d) DO UPDATE SET
    open = excluded.open, high = excluded.high, low = excluded.low,
    close = excluded.close, last = excluded.last, prev_close = excluded.prev_close,
    volume = excluded.volume, value = excluded.value, trades = excluded.trades
"""


def sync(con, ins_code):
    """Backfill if the instrument is new, otherwise pull only the recent days."""
    last = con.execute(
        "SELECT max(d) FROM daily WHERE ins_code = %s", (ins_code,)
    ).fetchone()[0]  # a datetime.date or None
    start = (last - timedelta(days=OVERLAP_DAYS)).isoformat() if last else None
    rows = fetch_daily(ins_code, start)
    with con.transaction():
        con.cursor().executemany(UPSERT, [{**r, "ins_code": ins_code} for r in rows])
    return len(rows)


# in main(): replace sqlite3.connect(DB_PATH) with
con = psycopg.connect(os.environ["DATABASE_URL"], autocommit=True)

قیمت تعدیل‌شده را چطور نگه دارم؟

قیمت تعدیل‌شده را مثل قیمت خام افزایشی ذخیره نکنید: هر رویداد تعدیل تازه، قیمت تعدیل‌شدهٔ همهٔ روزهای پیش از خودش را عوض می‌کند. دو راه درست دارید.

1. هنگام نیاز بگیرید. برای تحلیل، /daily?adjusted=true را مستقیم صدا بزنید و نتیجه را ذخیرهٔ بلندمدت نکنید.
2. رویدادها را دنبال کنید. تعداد رویدادهای /adjustments را کنار داده نگه دارید. اگر زیاد شد، کل سابقهٔ تعدیل‌شدهٔ آن نماد را دوباره بگیرید و جایگزین کنید.

این برنامه راه دوم را نشان می‌دهد. فایل کوچکی تعداد رویدادهای دیده‌شده را نگه می‌دارد و فقط وقتی رویداد تازه‌ای باشد سابقهٔ تعدیل‌شده را دوباره می‌گیرد:

"""Refresh a cached adjusted history only when a new adjustment event appears."""
import json
import os
import time
from pathlib import Path

import requests

BASE = "https://poolick.ir"
HEADERS = {"Authorization": f"Bearer {os.environ['ROSHA_KEY']}"}
INS_CODE = "46348559193224090"  # فولاد
STATE = Path(f"adjusted-{INS_CODE}.json")


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()


cached = json.loads(STATE.read_text()) if STATE.exists() else {"events": -1, "rows": []}
events = get(f"/v1/instruments/{INS_CODE}/adjustments")

if len(events) != cached["events"]:
    rows = get(
        f"/v1/instruments/{INS_CODE}/daily",
        params={"adjusted": "true", "limit": 5000},
    )["rows"]
    STATE.write_text(json.dumps({"events": len(events), "rows": rows}))
    print(f"refreshed: {len(events)} events, {len(rows)} adjusted rows")
else:
    print(f"unchanged: {len(events)} events, cache is still valid")

این نمونه تا ۵٬۰۰۰ ردیف آخر را نگه می‌دارد. برای نمادهای با سابقهٔ بلندتر، تابع fetch_daily برنامهٔ اول را با adjusted=true به کار ببرید.

صندوق‌ها و جدول‌های دیگر

همین الگو برای بقیهٔ endpointهای سابقه کار می‌کند؛ فقط کلید و زمان‌بندی فرق دارد.

داده	endpoint	کلید upsert	زمان اجرا (وقت تهران)
حقیقی و حقوقی	/v1/instruments/{ref}/client-type	(ins_code, d)	پس از ۱۸:۱۵، شنبه تا چهارشنبه
شاخص	/v1/indices/{ref}/daily	(ins_code, d)	پس از ۱۸:۱۵، شنبه تا چهارشنبه
سابقهٔ صندوق	/v1/funds/{ref}/history	(reg_no, d)	روزانه پس از ۲۲:۳۰؛ بازخوانی کامل هر هفته پس از جمعه ۰۲:۰۰
snapshot زیرصندوق	/v1/funds/{ref}/snapshots	(reg_no, group_id, d)	روزانه پس از ۲۲:۳۰

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