Skip to content

Secmaster

API endpoints for security master database queries.

Provides endpoints for querying publishers, datasets, and symbol coverage information.

router = APIRouter(prefix='/api/secmaster', tags=['secmaster']) module-attribute

api_secmaster_publishers(rtype=None) async

Return list of publishers, optionally filtered by rtype.

Source code in src/onesecondtrader/dashboard/routers/secmaster.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@router.get("/publishers")
async def api_secmaster_publishers(rtype: int | None = None) -> dict:
    """Return list of publishers, optionally filtered by rtype."""
    db_path = get_secmaster_path()
    if not os.path.exists(db_path):
        return {"publishers": []}
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    if rtype is not None:
        cursor.execute(
            "SELECT DISTINCT p.name FROM publishers p "
            "JOIN symbol_coverage sc ON p.publisher_id = sc.publisher_id "
            "WHERE sc.rtype = ? ORDER BY p.name",
            (rtype,),
        )
    else:
        cursor.execute("SELECT DISTINCT name FROM publishers ORDER BY name")
    publishers = [row[0] for row in cursor.fetchall()]
    conn.close()
    return {"publishers": publishers}

api_secmaster_datasets(name, rtype=None) async

Return datasets for a publisher, optionally filtered by rtype.

Source code in src/onesecondtrader/dashboard/routers/secmaster.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@router.get("/publishers/{name}/datasets")
async def api_secmaster_datasets(name: str, rtype: int | None = None) -> dict:
    """Return datasets for a publisher, optionally filtered by rtype."""
    db_path = get_secmaster_path()
    if not os.path.exists(db_path):
        return {"datasets": []}
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    if rtype is not None:
        cursor.execute(
            "SELECT DISTINCT p.publisher_id, p.dataset FROM publishers p "
            "JOIN symbol_coverage sc ON p.publisher_id = sc.publisher_id "
            "WHERE p.name = ? AND sc.rtype = ? ORDER BY p.dataset",
            (name, rtype),
        )
    else:
        cursor.execute(
            "SELECT publisher_id, dataset FROM publishers WHERE name = ? ORDER BY dataset",
            (name,),
        )
    datasets = [
        {"publisher_id": row[0], "dataset": row[1]} for row in cursor.fetchall()
    ]
    conn.close()
    return {"datasets": datasets}

api_secmaster_symbols_coverage(publisher_id=None, rtype=None) async

Return symbol coverage data, optionally filtered by publisher_id and rtype.

Source code in src/onesecondtrader/dashboard/routers/secmaster.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@router.get("/symbols_coverage")
async def api_secmaster_symbols_coverage(
    publisher_id: int | None = None, rtype: int | None = None
) -> dict:
    """Return symbol coverage data, optionally filtered by publisher_id and rtype."""
    db_path = get_secmaster_path()
    if not os.path.exists(db_path):
        return {"symbols": []}
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    if publisher_id is not None and rtype is not None:
        cursor.execute(
            "SELECT publisher_id, symbol, rtype, min_ts, max_ts FROM symbol_coverage "
            "WHERE publisher_id = ? AND rtype = ? ORDER BY symbol",
            (publisher_id, rtype),
        )
    elif publisher_id is not None:
        cursor.execute(
            "SELECT publisher_id, symbol, rtype, min_ts, max_ts FROM symbol_coverage "
            "WHERE publisher_id = ? ORDER BY symbol, rtype",
            (publisher_id,),
        )
    elif rtype is not None:
        cursor.execute(
            "SELECT publisher_id, symbol, rtype, min_ts, max_ts FROM symbol_coverage "
            "WHERE rtype = ? ORDER BY symbol",
            (rtype,),
        )
    else:
        cursor.execute(
            "SELECT publisher_id, symbol, rtype, min_ts, max_ts FROM symbol_coverage ORDER BY symbol, rtype"
        )
    symbols = [
        {
            "publisher_id": row[0],
            "symbol": row[1],
            "rtype": row[2],
            "min_ts": row[3],
            "max_ts": row[4],
        }
        for row in cursor.fetchall()
    ]
    conn.close()
    return {"symbols": symbols}