OHLC.dev editorialIDX

Build an IDX Stock Search and Morning Briefing Dashboard

This project demonstrates how to build an IDX Stock Search and Morning Briefing Dashboard using Python. The notebook connects to the Indonesia Stock Exchange API through RapidAPI, retrieves stock search results and Morning Briefing information, normalizes multiple JSON response structures, converts them into structured DataFrames, and generates a dashboard that summarizes both datasets for daily market monitoring.

July 22, 202611 min readRafatar
Build an IDX Stock Search and Morning Briefing Dashboard

Finding accurate company information is one of the first steps before analyzing a stock. However, investors also need to understand the broader market context through the latest daily market updates.

The Indonesia Stock Exchange API provides endpoints that can automate both tasks. One endpoint searches listed companies using a keyword, while another delivers the latest Morning Briefing published by the Indonesia Stock Exchange.

In this tutorial, we build an IDX Stock Search and Morning Briefing Dashboard using Python. The notebook authenticates through RapidAPI, retrieves stock search results together with Morning Briefing data, normalizes multiple JSON response structures, converts the responses into pandas DataFrames, and generates a dashboard summarizing both datasets in a single workflow.

The notebook is organized into five sections covering API configuration, secure API requests, response normalization, data presentation, and dashboard generation.



CELL 1 — Import Library dan Konfigurasi API

# ============================================================
# CELL 1 — IMPORT LIBRARY DAN KONFIGURASI API
# ============================================================

import requests
import pandas as pd
import json
import time
from datetime import datetime

# Masukkan RapidAPI Key Anda
API_KEY = "YOUR_RAPIDAPI_KEY"

BASE_URL = "https://indonesia-stock-exchange-idx.p.rapidapi.com"

HEADERS = {
    "Content-Type": "application/json",
    "x-rapidapi-key": API_KEY,
    "x-rapidapi-host": "indonesia-stock-exchange-idx.p.rapidapi.com"
}

# Parameter pencarian saham
KEYWORD = "BBCA"
SEARCH_TYPE = "company"
PAGE = 0

print("=" * 100)
print("KONFIGURASI IDX API")
print("=" * 100)
print(f"Keyword pencarian : {KEYWORD}")
print(f"Tipe pencarian    : {SEARCH_TYPE}")
print(f"Halaman           : {PAGE}")
print(f"Waktu eksekusi    : {datetime.now().strftime('%d-%m-%Y %H:%M:%S')}")

if API_KEY == "MASUKKAN_RAPIDAPI_KEY_ANDA":
    print("\n⚠️ Silakan masukkan RapidAPI Key pada variabel API_KEY.")
else:
    print("\n✅ Konfigurasi API berhasil disiapkan.")

Catatan: API Key pada notebook diganti menjadi YOUR_RAPIDAPI_KEY agar aman ketika dipublikasikan. Struktur kode lainnya tetap sama seperti file asli.

Explanation

The first cell initializes the Python environment and prepares the connection to the Indonesia Stock Exchange API through RapidAPI.

The notebook imports several libraries that will be used throughout the project. requests handles HTTP communication with the API, pandas converts JSON responses into structured DataFrames, json assists with debugging nested responses, time provides delays between API requests, and datetime records the execution timestamp displayed in the notebook.

Next, the notebook defines the API configuration. The RapidAPI key, API base URL, and HTTP headers are stored in dedicated variables so every request can reuse the same authentication settings.

The search configuration is also initialized. The notebook searches for companies using the keyword BBCA, sets the search type to company, and starts from the first result page.

Finally, the notebook prints the active configuration together with the execution time and verifies whether a valid API key has been provided before continuing to the next step.

CELL 2 — Fungsi Request API yang Aman

# ============================================================
# CELL 2 — FUNGSI REQUEST API YANG AMAN
# ============================================================

def request_api(endpoint, params=None, retries=2, timeout=30):
    """
    Mengambil data API dengan penanganan:
    - Timeout
    - Rate limit 429
    - Forbidden 403
    - Respons bukan JSON
    - Server error
    """

    url = f"{BASE_URL}{endpoint}"

    result = {
        "endpoint": endpoint,
        "status_code": None,
        "success": False,
        "data": None,
        "message": ""
    }

    for attempt in range(1, retries + 1):
        try:
            response = requests.get(
                url,
                headers=HEADERS,
                params=params,
                timeout=timeout
            )

            result["status_code"] = response.status_code

            print("-" * 100)
            print(f"Endpoint : {response.url}")
            print(f"Status   : {response.status_code}")
            print(f"Percobaan: {attempt}/{retries}")

            if response.status_code == 200:
                try:
                    result["data"] = response.json()
                    result["success"] = True
                    result["message"] = "Data berhasil diambil."
                    return result

                except ValueError:
                    result["message"] = "Respons API bukan format JSON."
                    return result

            elif response.status_code == 403:
                result["message"] = (
                    "Akses ditolak. Pastikan API Key benar dan Anda sudah "
                    "berlangganan API ini."
                )
                return result

            elif response.status_code == 429:
                result["message"] = "Rate limit tercapai. Terlalu banyak request."

                if attempt < retries:
                    print("⚠️ Menunggu sebelum mencoba kembali...")
                    time.sleep(3)
                    continue

                return result

            elif response.status_code >= 500:
                result["message"] = (
                    f"Server API mengalami gangguan "
                    f"({response.status_code})."
                )

                if attempt < retries:
                    time.sleep(2)
                    continue

                return result

            else:
                result["message"] = (
                    f"Request gagal dengan status {response.status_code}: "
                    f"{response.text[:300]}"
                )
                return result

        except requests.exceptions.Timeout:
            result["message"] = (
                f"Request mengalami timeout setelah {timeout} detik."
            )

            print(f"⚠️ {result['message']}")

            if attempt < retries:
                time.sleep(2)
                continue

        except requests.exceptions.ConnectionError:
            result["message"] = "Tidak dapat terhubung ke server API."

            if attempt < retries:
                time.sleep(2)
                continue

        except requests.exceptions.RequestException as error:
            result["message"] = f"Terjadi kesalahan request: {error}"
            return result

        except Exception as error:
            result["message"] = f"Terjadi kesalahan tidak terduga: {error}"
            return result

    return result

print("✅ Fungsi request_api berhasil dibuat.")

Explanation

The second cell builds a reusable request function that handles communication with the Indonesia Stock Exchange API.

Instead of sending requests directly throughout the notebook, every API call passes through the request_api() function. This keeps the request logic centralized and ensures consistent error handling across all endpoints.

The function returns a structured dictionary containing the endpoint, HTTP status code, success flag, parsed JSON data, and a descriptive message.

Several common API scenarios are handled automatically:

  • HTTP 200 returns the JSON response successfully.

  • HTTP 403 indicates that the API subscription or credentials do not permit access.

  • HTTP 429 indicates that the RapidAPI request limit has been reached, triggering a retry after a short delay.

  • HTTP 5xx responses are treated as temporary server errors and are retried when possible.

The function also catches timeout exceptions, connection failures, invalid JSON responses, and unexpected request errors, allowing the notebook to continue running without terminating unexpectedly.


CELL 3 — Mengambil Search Stocks dan Morning Briefing

# ============================================================
# CELL 3 — REQUEST SEARCH STOCKS DAN MORNING BRIEFING
# ============================================================

# Endpoint 1: Search Stocks
search_result = request_api(
    endpoint="/api/main/search",
    params={
        "type": SEARCH_TYPE,
        "keyword": KEYWORD,
        "page": PAGE
    }
)

# Beri jeda agar mengurangi kemungkinan rate limit
time.sleep(1)

# Endpoint 2: Morning Briefing
morning_result = request_api(
    endpoint="/api/main/morning-briefing"
)

# Menyimpan respons mentah
search_response = search_result.get("data")
morning_response = morning_result.get("data")

print("\n" + "=" * 100)
print("STATUS PENGAMBILAN DATA")
print("=" * 100)

print("\n1. SEARCH STOCKS")
print(f"Status HTTP : {search_result.get('status_code')}")
print(f"Berhasil    : {search_result.get('success')}")
print(f"Keterangan : {search_result.get('message')}")

print("\n2. MORNING BRIEFING")
print(f"Status HTTP : {morning_result.get('status_code')}")
print(f"Berhasil    : {morning_result.get('success')}")
print(f"Keterangan : {morning_result.get('message')}")

print("\n" + "=" * 100)
print("DEBUG STRUKTUR RESPONS")
print("=" * 100)

print("\nSearch Stocks:")
print("Tipe respons :", type(search_response).__name__)

if isinstance(search_response, dict):
    print("Key utama   :", list(search_response.keys()))

print("\nMorning Briefing:")
print("Tipe respons :", type(morning_response).__name__)

if isinstance(morning_response, dict):
    print("Key utama   :", list(morning_response.keys()))

Explanation

The third cell retrieves data from the two API endpoints used in the project.

The notebook first requests Search Stocks by sending the configured keyword, search type, and page number to the /api/main/search endpoint.

After the first request completes, the notebook waits for one second before sending another request. This short delay helps reduce the possibility of reaching the RapidAPI rate limit when multiple endpoints are accessed consecutively.

The second request retrieves the latest Morning Briefing using the /api/main/morning-briefing endpoint.

Once both requests finish, the notebook extracts the JSON content from each response and stores it separately as search_response and morning_response.

To help verify that both endpoints responded correctly, the notebook prints a status summary showing the HTTP status code, request success flag, and descriptive message returned by the reusable request function.

Finally, a debugging section displays the response type and the top-level JSON keys for each endpoint whenever the response is a dictionary. This information becomes particularly useful when the API response structure changes and needs to be normalized in the following cell.


CELL 4 — Normalisasi dan Pemrosesan Data

# ============================================================
# CELL 4 — NORMALISASI DAN PEMROSESAN DATA
# ============================================================

def find_records(data):
    """
    Mencari kumpulan record berbentuk list dari berbagai
    kemungkinan struktur respons API.
    """

    if data is None:
        return []

    if isinstance(data, list):
        return data

    if not isinstance(data, dict):
        return []

    possible_keys = [
        "data",
        "results",
        "result",
        "items",
        "records",
        "companies",
        "stocks",
        "searchResults",
        "content",
        "list"
    ]

    for key in possible_keys:
        value = data.get(key)

        if isinstance(value, list):
            return value

        if isinstance(value, dict):
            nested_records = find_records(value)

            if nested_records:
                return nested_records

    # Memeriksa seluruh value jika key respons berbeda
    for value in data.values():
        if isinstance(value, list) and value:
            if isinstance(value[0], dict):
                return value

    # Menganggap dictionary sebagai satu record
    excluded_keys = {"success", "message", "status", "code"}

    if any(key not in excluded_keys for key in data.keys()):
        return [data]

    return []


def get_first_value(item, possible_keys, default="-"):
    """
    Mengambil nilai pertama yang tersedia dari beberapa kemungkinan key.
    """

    if not isinstance(item, dict):
        return default

    for key in possible_keys:
        value = item.get(key)

        if value not in [None, "", [], {}]:
            if isinstance(value, (dict, list)):
                return json.dumps(value, ensure_ascii=False)

            return value

    return default


def flatten_briefing(data):
    """
    Mengubah Morning Briefing menjadi daftar informasi
    yang lebih mudah ditampilkan.
    """

    records = find_records(data)
    briefing_items = []

    for index, item in enumerate(records, start=1):
        if isinstance(item, dict):
            title = get_first_value(
                item,
                [
                    "title",
                    "headline",
                    "name",
                    "subject",
                    "category",
                    "topic",
                    "type"
                ],
                f"Informasi {index}"
            )

            description = get_first_value(
                item,
                [
                    "description",
                    "summary",
                    "content",
                    "text",
                    "message",
                    "briefing",
                    "analysis",
                    "value"
                ],
                "-"
            )

            sentiment = get_first_value(
                item,
                [
                    "sentiment",
                    "status",
                    "signal",
                    "impact",
                    "trend"
                ],
                "-"
            )

            briefing_items.append({
                "Judul": title,
                "Deskripsi": description,
                "Sentimen": sentiment
            })

        elif isinstance(item, str):
            briefing_items.append({
                "Judul": f"Informasi {index}",
                "Deskripsi": item,
                "Sentimen": "-"
            })

    # Jika respons dictionary tidak memiliki list yang jelas
    if not briefing_items and isinstance(data, dict):
        main_data = data.get("data", data)

        if isinstance(main_data, dict):
            for key, value in main_data.items():
                if key.lower() not in ["success", "status", "code"]:
                    briefing_items.append({
                        "Judul": str(key).replace("_", " ").title(),
                        "Deskripsi": (
                            json.dumps(value, ensure_ascii=False)
                            if isinstance(value, (dict, list))
                            else value
                        ),
                        "Sentimen": "-"
                    })

    return briefing_items


# ==========================
# NORMALISASI SEARCH STOCKS
# ==========================

search_records = find_records(search_response)
search_clean = []

for item in search_records:
    if not isinstance(item, dict):
        continue

    search_clean.append({
        "Kode": get_first_value(
            item,
            ["symbol", "code", "ticker", "stockCode", "stock_code"]
        ),
        "Perusahaan": get_first_value(
            item,
            [
                "company_name",
                "companyName",
                "name",
                "company",
                "description"
            ]
        ),
        "Sektor": get_first_value(
            item,
            ["sector", "sector_name", "sectorName"]
        ),
        "Industri": get_first_value(
            item,
            ["industry", "industry_name", "subSector", "sub_sector"]
        ),
        "Jenis": get_first_value(
            item,
            ["type", "category", "instrument_type"]
        )
    })


df_search = pd.DataFrame(search_clean)

if not df_search.empty:
    df_search = df_search.drop_duplicates().reset_index(drop=True)


# ==========================
# NORMALISASI MORNING BRIEFING
# ==========================

briefing_clean = flatten_briefing(morning_response)
df_briefing = pd.DataFrame(briefing_clean)

if not df_briefing.empty:
    df_briefing = df_briefing.drop_duplicates().reset_index(drop=True)


print("=" * 100)
print("HASIL NORMALISASI DATA")
print("=" * 100)
print(f"Search Stocks Records    : {len(df_search)}")
print(f"Morning Briefing Records : {len(df_briefing)}")

print("\nKolom Search Stocks:")
print(list(df_search.columns) if not df_search.empty else "Tidak ada kolom.")

print("\nKolom Morning Briefing:")
print(list(df_briefing.columns) if not df_briefing.empty else "Tidak ada kolom.")

Explanation

Cell 4 processes the raw responses from both endpoints into consistent pandas DataFrames.

The find_records() function searches for record collections inside several possible JSON wrappers, including data, results, items, companies, stocks, and content. It also checks nested dictionaries recursively and can treat a single dictionary as one record.

The get_first_value() function searches several possible field names and returns the first usable value. This allows the notebook to handle variations such as symbol, code, ticker, or stockCode without producing a KeyError.

For Morning Briefing, flatten_briefing() converts the response into three consistent fields:

  • Judul

  • Deskripsi

  • Sentimen

The Search Stocks response is converted into a DataFrame containing stock code, company, sector, industry, and instrument type. Duplicate records are removed from both DataFrames before the notebook prints the normalization summary and available columns.

CELL 5 — Dashboard IDX Market

# ============================================================
# CELL 5 — DASHBOARD IDX MARKET
# ============================================================

def shorten_text(value, max_length=250):
    """Memotong teks panjang agar output tetap rapi."""

    if value is None:
        return "-"

    text = str(value).strip()

    if not text:
        return "-"

    if len(text) > max_length:
        return text[:max_length].rstrip() + "..."

    return text


print("=" * 100)
print("IDX MARKET DASHBOARD")
print("=" * 100)
print(f"Keyword Saham : {KEYWORD}")
print(f"Waktu Update  : {datetime.now().strftime('%d-%m-%Y %H:%M:%S')}")

# ============================================================
# SEARCH STOCKS
# ============================================================

print("\n")
print("🔎 SEARCH STOCKS")
print("-" * 100)

if not df_search.empty:
    print(f"Jumlah hasil ditemukan: {len(df_search)}\n")

    for index, row in df_search.head(10).iterrows():
        print(f"{index + 1:02d}. {row.get('Kode', '-')}")
        print(f"    Perusahaan : {row.get('Perusahaan', '-')}")
        print(f"    Sektor     : {row.get('Sektor', '-')}")
        print(f"    Industri   : {row.get('Industri', '-')}")
        print(f"    Jenis      : {row.get('Jenis', '-')}")
        print("-" * 100)

else:
    print("Tidak ada data hasil pencarian saham.")

    if not search_result.get("success"):
        print(f"Keterangan: {search_result.get('message')}")


# ============================================================
# MORNING BRIEFING
# ============================================================

print("\n")
print("🌅 MORNING BRIEFING")
print("-" * 100)

if not df_briefing.empty:
    print(f"Jumlah informasi: {len(df_briefing)}\n")

    for index, row in df_briefing.head(10).iterrows():
        title = shorten_text(row.get("Judul", "-"), 100)
        description = shorten_text(row.get("Deskripsi", "-"), 350)
        sentiment = shorten_text(row.get("Sentimen", "-"), 50)

        print(f"{index + 1:02d}. {title}")
        print(f"    Sentimen   : {sentiment}")
        print(f"    Informasi  : {description}")
        print("-" * 100)

else:
    print("Tidak ada data Morning Briefing.")

    if not morning_result.get("success"):
        print(f"Keterangan: {morning_result.get('message')}")


# ============================================================
# RINGKASAN
# ============================================================

print("\n")
print("=" * 100)
print("RINGKASAN")
print("=" * 100)

search_status = (
    "Data berhasil diambil"
    if search_result.get("success")
    else search_result.get("message", "Request gagal")
)

morning_status = (
    "Data berhasil diambil"
    if morning_result.get("success")
    else morning_result.get("message", "Request gagal")
)

print(f"🔎 Keyword pencarian       : {KEYWORD}")
print(f"📊 Jumlah hasil saham      : {len(df_search)}")
print(f"🌅 Jumlah informasi pagi   : {len(df_briefing)}")
print(f"✅ Status Search Stocks    : {search_status}")
print(f"✅ Status Morning Briefing : {morning_status}")

if not df_search.empty:
    kode_ditemukan = ", ".join(
        df_search["Kode"]
        .astype(str)
        .replace("-", pd.NA)
        .dropna()
        .unique()
        .tolist()[:10]
    )

    print(
        f"🏢 Saham ditemukan         : "
        f"{kode_ditemukan if kode_ditemukan else '-'}"
    )

if not search_result.get("success") or not morning_result.get("success"):
    print("\n⚠️ Periksa API Key, paket langganan, dan batas penggunaan RapidAPI.")

print("=" * 100)

Explanation

Cell 5 generates the final text-based IDX market dashboard.

The shorten_text() function limits long Morning Briefing content so the dashboard remains readable. It also handles empty or missing values by replacing them with a hyphen.

The Search Stocks section displays up to ten results, including the stock code, company name, sector, industry, and instrument type.

The Morning Briefing section also displays up to ten records. Each record contains its title, sentiment, and shortened description.

The final Ringkasan section reports the active search keyword, number of stock results, number of Morning Briefing records, and request status for both endpoints. When stock results are available, the detected stock codes are combined into one summary line.

If either request fails, the notebook reminds users to check their API key, RapidAPI subscription, and usage limit.

Conclusion

The completed notebook combines company search and daily market information into one practical IDX dashboard.

It safely handles API errors, recognizes different JSON structures, standardizes important fields, removes duplicate records, and presents the results in an organized output. This makes the project useful as a foundation for an automated Indonesian stock-search and Morning Briefing monitoring tool.