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

> نسخهٔ محلی سابقهٔ روزانهٔ بورس با API روشا؛ بارگیری اولیه با صفحه‌بندی تاریخ، کار روزانه پس از ۱۸:۱۵ تهران و upsert در SQLite یا PostgreSQL.
> Source: https://poolick.ir/docs/guides/daily-sync/

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

## طرح کلی چیست؟

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

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

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

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

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

## کد کامل با SQLite

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

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

Run once to backfill, then every trading day after 18:15 Asia/Tehran.
"""

from datetime import date, timedelta

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 جلوتر است):

```bash
# 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`](/docs/api/calendar/) را ببینید.

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

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

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

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

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

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

```sql
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]"`) این سه بخش برنامهٔ بالا را جایگزین کنید:

```python

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`](/docs/api/adjustments/) را کنار داده نگه دارید. اگر زیاد شد، کل سابقهٔ تعدیل‌شدهٔ آن نماد را دوباره بگیرید و جایگزین کنید.

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

```python
"""Refresh a cached adjusted history only when a new adjustment event appears."""

from pathlib import Path

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)` | روزانه پس از ۲۲:۳۰ |

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