"""
Exodus REST API — Main Stockbit API client.
Handles user, social, watchlist, chart, company data, and chat endpoints.
"""
from typing import Optional, List
import requests
from .auth import AuthManager
from .utils import RateLimiter, retry
class ExodusAPI:
BASE = "https://exodus.stockbit.com"
def __init__(self, auth: AuthManager, session: requests.Session, rate_limiter: RateLimiter):
self._auth = auth
self._session = session
self._rl = rate_limiter
# ── User Profile ──────────────────────────────────────
@retry()
def get_profile(self) -> dict:
return self._get("/user/profile")
@retry()
def get_profile_by_username(self, username: str) -> dict:
return self._get(f"/user/profile/{username}")
@retry()
def edit_profile(self, **kwargs) -> dict:
return self._put("/user/profile", json=kwargs)
# ── Social / Stream ───────────────────────────────────
@retry()
def get_stream(self) -> dict:
return self._get("/stream/get")
@retry()
def get_trending_posts(self, date: Optional[str] = None) -> dict:
params = {"date": date} if date else {}
return self._get("/stream/trending", params=params)
@retry()
def get_stream_v3(self) -> dict:
return self._get("/stream/v3")
@retry()
def get_stream_by_symbol(self, symbol: str) -> dict:
return self._get(f"/stream/v3/symbol/{symbol}")
# ── Social Graph ──────────────────────────────────────
@retry()
def get_my_social_info(self) -> dict:
return self._get("/usergraph/socialinfo/user/me")
@retry()
def get_user_social_info(self, username: str) -> dict:
return self._get(f"/usergraph/socialinfo/{username}")
# ── Search ────────────────────────────────────────────
@retry()
def search(self, keyword: str, page: int = 0,
search_type: str = "all") -> dict:
return self._get("/search", params={
"keyword": keyword, "page": page, "type": search_type,
})
# ── Watchlist ─────────────────────────────────────────
@retry()
def get_watchlists(self) -> dict:
return self._get("/watchlist")
@retry()
def watchlist_list(self) -> dict:
return self._post("/watchlist/list")
@retry()
def watchlist_create(self) -> dict:
return self._post("/watchlist/create")
@retry()
def watchlist_add(self, symbol: str) -> dict:
return self._post("/watchlist/add", json={"symbol": symbol})
@retry()
def watchlist_remove(self, symbol: str) -> dict:
return self._post("/watchlist/remove", json={"symbol": symbol})
@retry()
def get_watchlist_metrics(self) -> dict:
return self._get("/watchlist/metric")
# ── Chart Data ────────────────────────────────────────
@retry()
def get_daily_price(self, symbol: str, from_date: Optional[str] = None,
to_date: Optional[str] = None) -> dict:
params = {}
if from_date:
params["from"] = from_date
if to_date:
params["to"] = to_date
return self._get(f"/chartbit/{symbol}/price/daily", params=params)
@retry()
def get_intraday_price(self, symbol: str, from_ts: Optional[int] = None,
to_ts: Optional[int] = None) -> dict:
params = {}
if from_ts:
params["from"] = from_ts
if to_ts:
params["to"] = to_ts
return self._get(f"/chartbit/{symbol}/price/intraday", params=params)
# ── Company Info ──────────────────────────────────────
@retry()
def get_company_info(self, symbol: str) -> dict:
return self._get(f"/emitten/non-login/{symbol}/info")
@retry()
def get_trending_companies(self) -> dict:
return self._get("/emitten/trending")
@retry()
def get_index_info(self) -> dict:
return self._get("/emitten/IHSG/info")
@retry()
def get_broker_activity(self, symbol: str, page: int = 1,
limit: int = 20) -> dict:
return self._get("/broker/activity", params={
"symbol": symbol, "page": page, "limit": limit,
})
@retry()
def get_broker_list(self) -> dict:
return self._get("/broker/list")
@retry()
def get_orderbook(self, symbol: str) -> dict:
return self._get(f"/orderbook/companies/{symbol}")
# ── Key Stats / Ratios ─────────────────────────────────
@retry()
def get_key_ratios(self, symbol: str, year_limit: int = 10) -> dict:
return self._get(f"/keystats/ratio/v1/{symbol}",
params={"year_limit": year_limit})
# ── Price Performance ──────────────────────────────────
@retry()
def get_price_performance(self, symbol: str) -> dict:
return self._get(f"/company-price-feed/price-performance/{symbol}")
# ── Charts (daily) ─────────────────────────────────────
@retry()
def get_daily_chart(self, symbol: str, timeframe: str = "today") -> dict:
return self._get(f"/charts/{symbol}/daily",
params={"timeframe": timeframe})
# ── Top Brokers ────────────────────────────────────────
@retry()
def get_top_brokers(self, sort: str = "TB_SORT_BY_TOTAL_VALUE",
order: str = "ORDER_BY_DESC",
period: str = "TB_PERIOD_LAST_1_DAY",
market_type: str = "MARKET_TYPE_ALL",
eod_only: bool = True) -> dict:
return self._get("/order-trade/broker/top", params={
"sort": sort,
"order": order,
"period": period,
"market_type": market_type,
"eod_only": "true" if eod_only else "false",
})
# ── Financial Reports ──────────────────────────────────
@retry()
def get_financial_report(self, symbol: str, data_type: int = 1,
report_type: int = 1,
statement_type: int = 1) -> dict:
return self._get("/findata-view/company/financial", params={
"symbol": symbol,
"data_type": data_type,
"report_type": report_type,
"statement_type": statement_type,
})
# ── Analyst ────────────────────────────────────────────
@retry()
def get_analyst(self, symbol: str, build_id: str = "") -> dict:
path = f"/_next/data/{build_id}/symbol/{symbol}/analyst.json"
return self._get(f"https://stockbit.com{path}",
params={"symbol": symbol, "param": "analyst"})
# ── Insider / Major Holder ─────────────────────────────
@retry()
def get_major_holder(self, symbol: str, page: int = 1,
limit: int = 20,
action_type: str = "ACTION_TYPE_UNSPECIFIED",
source_type: str = "SOURCE_TYPE_UNSPECIFIED",
insider: str = "",
date_start: str = "",
date_end: str = "") -> dict:
params = {
"symbols": symbol,
"page": page,
"limit": limit,
"action_type": action_type,
"source_type": source_type,
}
if insider:
params["insider"] = insider
if date_start:
params["date_start"] = date_start
if date_end:
params["date_end"] = date_end
return self._get("/insider/company/majorholder", params=params)
# ── Corporate Action ───────────────────────────────────
@retry()
def get_corporate_action(self, symbol: str, limit: int = 30) -> dict:
return self._get(f"/corpaction/{symbol}", params={"limit": limit})
# ── Company Profile ────────────────────────────────────
@retry()
def get_company_profile(self, symbol: str) -> dict:
return self._get(f"/emitten/{symbol}/profile")
# ── Seasonality ────────────────────────────────────────
@retry()
def get_seasonality(self, symbol: str, year: int = 2026,
back_year: int = 5) -> dict:
return self._get(f"/company-price-feed/seasonality/{symbol}",
params={"year": year, "back_year": back_year})
# ── Chartbit Price Data ────────────────────────────────
@retry()
def get_chartbit_price(self, symbol: str, from_date: str = "",
to_date: str = "", limit: int = 0) -> dict:
return self._get(f"/chartbit/{symbol}/price/daily",
params={"from": from_date, "to": to_date,
"limit": limit})
# ── Sectors ───────────────────────────────────────────
@retry()
def get_sectors(self) -> dict:
return self._get("/sectors")
@retry()
def get_subsectors(self) -> dict:
return self._get("/subsectors")
# ── Notifications ─────────────────────────────────────
@retry()
def get_notifications(self) -> dict:
return self._get("/notification")
@retry()
def get_unread_count(self) -> dict:
return self._get("/count/unread")
@retry()
def mark_notifications_read(self) -> dict:
return self._post("/notification/read")
# ── Chat ──────────────────────────────────────────────
@retry()
def get_chat_rooms(self) -> dict:
return self._get("/chat/v2/rooms")
@retry()
def get_chat_room_detail(self, room_id: str) -> dict:
return self._get(f"/chat/v2/rooms/{room_id}")
@retry()
def get_chat_messages(self, room_id: str, limit: int = 50,
before_id: Optional[str] = None) -> dict:
params = {"limit": limit}
if before_id:
params["before_id"] = before_id
return self._get(f"/chat/rooms/{room_id}/messages", params=params)
@retry()
def send_chat_message(self, room_id: str, text: str) -> dict:
return self._post(f"/chat/rooms/{room_id}/messages",
json={"text": text})
# ── Search People / Company ───────────────────────────
@retry()
def search_people(self, keyword: str, page: int = 0) -> dict:
return self._get("/search", params={
"keyword": keyword, "page": page, "type": "users",
})
# ── Internals ────────────────────────────────────────
def _get(self, path: str, **kwargs) -> dict:
wait = self._rl.acquire()
if wait > 0:
import time
time.sleep(wait)
url = path if path.startswith("http") else f"{self.BASE}{path}"
resp = self._session.get(url, **kwargs)
resp.raise_for_status()
return resp.json()
def _post(self, path: str, **kwargs) -> dict:
wait = self._rl.acquire()
if wait > 0:
import time
time.sleep(wait)
resp = self._session.post(f"{self.BASE}{path}", **kwargs)
resp.raise_for_status()
return resp.json()
def _put(self, path: str, **kwargs) -> dict:
wait = self._rl.acquire()
if wait > 0:
import time
time.sleep(wait)
resp = self._session.put(f"{self.BASE}{path}", **kwargs)
resp.raise_for_status()
return resp.json()
def _delete(self, path: str, **kwargs) -> dict:
wait = self._rl.acquire()
if wait > 0:
import time
time.sleep(wait)
resp = self._session.delete(f"{self.BASE}{path}", **kwargs)
resp.raise_for_status()
return resp.json()
Stockbit Research
research
Playground
@hansputera