Skip to content

Lookup Engine Reference

ISIN Lookup Table

tickertape.lookup.table.ISINLookupTable

Persistent SQLite store mapping mutual fund ISINs to TickerTape records and categorization.

Source code in src/tickertape/lookup/table.py 
 39
 40
 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
 66
 67
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
class ISINLookupTable:
    """Persistent SQLite store mapping mutual fund ISINs to TickerTape records and categorization."""

    def __init__(
        self,
        db_path: Optional[Path | str] = None,
        cache_dir: Path | str = DEFAULT_CACHE_DIR,
    ) -> None:
        if db_path:
            self.db_path = Path(db_path)
        else:
            self.db_path = Path(cache_dir) / "isin_lookup.db"

        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self._init_db()

    @contextmanager
    def _get_connection(self) -> Generator[sqlite3.Connection, None, None]:
        """Create and yield a configured sqlite3 connection, ensuring it is closed."""
        conn = sqlite3.connect(str(self.db_path), timeout=30.0)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
        finally:
            conn.close()

    def _init_db(self) -> None:
        """Initialize database schema, migrations, and indexes."""
        with self._get_connection() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS mf_isin_lookup (
                    isin TEXT PRIMARY KEY,
                    record_id TEXT NOT NULL,
                    slug TEXT NOT NULL,
                    name TEXT NOT NULL,
                    amc TEXT,
                    amc_code TEXT,
                    sector TEXT,
                    subsector TEXT,
                    fund_type TEXT,
                    fund_class TEXT,
                    plan TEXT,
                    option TEXT,
                    risk_level TEXT,
                    benchmark TEXT,
                    url TEXT NOT NULL,
                    nav REAL,
                    updated_at TEXT NOT NULL
                )
                """)

            # Auto-migration for existing databases
            cursor = conn.execute("PRAGMA table_info(mf_isin_lookup)")
            existing_cols = {row["name"] for row in cursor.fetchall()}
            for col_name, col_type in EXPECTED_COLUMNS.items():
                if col_name not in existing_cols:
                    conn.execute(
                        f"ALTER TABLE mf_isin_lookup ADD COLUMN {col_name} {col_type}"
                    )
                    logger.info(
                        "Migrated SQLite lookup table: added column '%s'", col_name
                    )

            # Indexes for fast grouping and peer clustering
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_lookup_record_id ON mf_isin_lookup(record_id)"
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_lookup_sector ON mf_isin_lookup(sector)"
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_lookup_subsector ON mf_isin_lookup(subsector)"
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_lookup_benchmark ON mf_isin_lookup(benchmark)"
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_lookup_amc ON mf_isin_lookup(amc)"
            )
            conn.commit()

    def get(self, isin: str) -> Optional[ISINMapping]:
        """Fetch mapping for a specific ISIN."""
        clean_isin = isin.strip().upper()
        with self._get_connection() as conn:
            cursor = conn.execute(
                "SELECT * FROM mf_isin_lookup WHERE isin = ?", (clean_isin,)
            )
            row = cursor.fetchone()
            if row:
                return ISINMapping.model_validate(dict(row))
        return None

    def get_batch(self, isins: list[str]) -> dict[str, ISINMapping]:
        """Fetch mappings for multiple ISINs at once."""
        clean_isins = [i.strip().upper() for i in isins if i.strip()]
        if not clean_isins:
            return {}

        placeholders = ",".join("?" for _ in clean_isins)
        result: dict[str, ISINMapping] = {}

        with self._get_connection() as conn:
            cursor = conn.execute(
                f"SELECT * FROM mf_isin_lookup WHERE isin IN ({placeholders})",
                clean_isins,
            )
            for row in cursor.fetchall():
                mapping = ISINMapping.model_validate(dict(row))
                result[mapping.isin] = mapping

        return result

    def get_by_record_id(self, record_id: str) -> Optional[ISINMapping]:
        """Fetch mapping by TickerTape record_id (e.g. 'M_QUNG')."""
        clean_id = record_id.strip()
        with self._get_connection() as conn:
            cursor = conn.execute(
                "SELECT * FROM mf_isin_lookup WHERE record_id = ?", (clean_id,)
            )
            row = cursor.fetchone()
            if row:
                return ISINMapping.model_validate(dict(row))
        return None

    def find_funds(
        self,
        *,
        sector: Optional[str] = None,
        subsector: Optional[str] = None,
        fund_type: Optional[str] = None,
        plan: Optional[str] = None,
        option: Optional[str] = None,
        risk_level: Optional[str] = None,
        benchmark: Optional[str] = None,
        amc: Optional[str] = None,
        limit: int = 100,
    ) -> list[ISINMapping]:
        """Gather mutual funds sharing similar classification attributes (peers).

        Useful for AI agents to compare funds in the same category or benchmark.
        """
        clauses: list[str] = []
        params: list[Any] = []

        if sector:
            clauses.append("LOWER(sector) = LOWER(?)")
            params.append(sector)
        if subsector:
            clauses.append("LOWER(subsector) = LOWER(?)")
            params.append(subsector)
        if fund_type:
            clauses.append("LOWER(fund_type) = LOWER(?)")
            params.append(fund_type)
        if plan:
            clauses.append("LOWER(plan) = LOWER(?)")
            params.append(plan)
        if option:
            clauses.append("LOWER(option) = LOWER(?)")
            params.append(option)
        if risk_level:
            clauses.append("LOWER(risk_level) = LOWER(?)")
            params.append(risk_level)
        if benchmark:
            clauses.append("LOWER(benchmark) = LOWER(?)")
            params.append(benchmark)
        if amc:
            clauses.append("LOWER(amc) LIKE LOWER(?)")
            params.append(f"%{amc}%")

        where_sql = f"WHERE {' AND '.join(clauses)}" if clauses else ""
        query = f"SELECT * FROM mf_isin_lookup {where_sql} ORDER BY name ASC LIMIT ?"
        params.append(limit)

        with self._get_connection() as conn:
            cursor = conn.execute(query, params)
            return [ISINMapping.model_validate(dict(row)) for row in cursor.fetchall()]

    def get_peers(
        self,
        isin: str,
        match_plan: bool = True,
        match_option: bool = True,
        limit: int = 20,
    ) -> list[ISINMapping]:
        """Get peer mutual funds in the same subsector/category as the given ISIN.

        Args:
            isin: Source fund ISIN.
            match_plan: If True, matches plan (e.g. Direct with Direct).
            match_option: If True, matches option (e.g. Growth with Growth).
            limit: Maximum peers to return.
        """
        fund = self.get(isin)
        if not fund or not fund.subsector:
            return []

        clauses = ["isin != ?", "LOWER(subsector) = LOWER(?)"]
        params: list[Any] = [fund.isin, fund.subsector]

        if match_plan and fund.plan:
            clauses.append("LOWER(plan) = LOWER(?)")
            params.append(fund.plan)
        if match_option and fund.option:
            clauses.append("LOWER(option) = LOWER(?)")
            params.append(fund.option)

        query = f"SELECT * FROM mf_isin_lookup WHERE {' AND '.join(clauses)} ORDER BY name ASC LIMIT ?"
        params.append(limit)

        with self._get_connection() as conn:
            cursor = conn.execute(query, params)
            return [ISINMapping.model_validate(dict(row)) for row in cursor.fetchall()]

    def get_all_record_ids(self) -> set[str]:
        """Return a set of all indexed TickerTape record_ids."""
        with self._get_connection() as conn:
            cursor = conn.execute("SELECT record_id FROM mf_isin_lookup")
            return {row[0] for row in cursor.fetchall()}

    def get_all_isins(self) -> set[str]:
        """Return a set of all indexed ISINs."""
        with self._get_connection() as conn:
            cursor = conn.execute("SELECT isin FROM mf_isin_lookup")
            return {row[0] for row in cursor.fetchall()}

    def upsert(self, mapping: ISINMapping) -> None:
        """Insert or update a single ISIN mapping."""
        self.upsert_batch([mapping])

    def upsert_batch(self, mappings: list[ISINMapping]) -> None:
        """Insert or update multiple ISIN mappings in a single atomic transaction."""
        if not mappings:
            return

        query = """
        INSERT INTO mf_isin_lookup (
            isin, record_id, slug, name, amc, amc_code, sector, subsector,
            fund_type, fund_class, plan, option, risk_level, benchmark, url, nav, updated_at
        ) VALUES (
            :isin, :record_id, :slug, :name, :amc, :amc_code, :sector, :subsector,
            :fund_type, :fund_class, :plan, :option, :risk_level, :benchmark, :url, :nav, :updated_at
        )
        ON CONFLICT(isin) DO UPDATE SET
            record_id = excluded.record_id,
            slug = excluded.slug,
            name = excluded.name,
            amc = excluded.amc,
            amc_code = excluded.amc_code,
            sector = excluded.sector,
            subsector = excluded.subsector,
            fund_type = excluded.fund_type,
            fund_class = excluded.fund_class,
            plan = excluded.plan,
            option = excluded.option,
            risk_level = excluded.risk_level,
            benchmark = excluded.benchmark,
            url = excluded.url,
            nav = excluded.nav,
            updated_at = excluded.updated_at
        """
        payloads = [m.model_dump() for m in mappings]
        with self._get_connection() as conn:
            conn.executemany(query, payloads)
            conn.commit()

        logger.debug("Upserted %d mappings into SQLite lookup table", len(mappings))

    def count(self) -> int:
        """Return total number of indexed mappings in lookup table."""
        with self._get_connection() as conn:
            cursor = conn.execute("SELECT COUNT(*) FROM mf_isin_lookup")
            return cursor.fetchone()[0]

    def all(self) -> list[ISINMapping]:
        """Return all mappings in the database."""
        with self._get_connection() as conn:
            cursor = conn.execute("SELECT * FROM mf_isin_lookup ORDER BY name ASC")
            return [ISINMapping.model_validate(dict(row)) for row in cursor.fetchall()]

    def clear(self) -> None:
        """Clear all entries from the lookup table."""
        with self._get_connection() as conn:
            conn.execute("DELETE FROM mf_isin_lookup")
            conn.commit()

    def export_json(self, filepath: Path | str) -> Path:
        """Export the full lookup table to a JSON file."""
        target = Path(filepath)
        target.parent.mkdir(parents=True, exist_ok=True)
        records = [m.model_dump() for m in self.all()]
        with open(target, "w", encoding="utf-8") as f:
            json.dump(records, f, indent=2)
        return target

    def export_csv(self, filepath: Path | str) -> Path:
        """Export the full lookup table to a CSV file."""
        target = Path(filepath)
        target.parent.mkdir(parents=True, exist_ok=True)
        records = [m.model_dump() for m in self.all()]
        if not records:
            fieldnames = list(ISINMapping.model_fields.keys())
        else:
            fieldnames = list(records[0].keys())

        with open(target, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(records)

        return target

all 

all() -> list[ISINMapping]

Return all mappings in the database.

Source code in src/tickertape/lookup/table.py 
313
314
315
316
317
def all(self) -> list[ISINMapping]:
    """Return all mappings in the database."""
    with self._get_connection() as conn:
        cursor = conn.execute("SELECT * FROM mf_isin_lookup ORDER BY name ASC")
        return [ISINMapping.model_validate(dict(row)) for row in cursor.fetchall()]

clear 

clear() -> None

Clear all entries from the lookup table.

Source code in src/tickertape/lookup/table.py 
319
320
321
322
323
def clear(self) -> None:
    """Clear all entries from the lookup table."""
    with self._get_connection() as conn:
        conn.execute("DELETE FROM mf_isin_lookup")
        conn.commit()

count 

count() -> int

Return total number of indexed mappings in lookup table.

Source code in src/tickertape/lookup/table.py 
307
308
309
310
311
def count(self) -> int:
    """Return total number of indexed mappings in lookup table."""
    with self._get_connection() as conn:
        cursor = conn.execute("SELECT COUNT(*) FROM mf_isin_lookup")
        return cursor.fetchone()[0]

export_csv 

export_csv(filepath: Path | str) -> Path

Export the full lookup table to a CSV file.

Source code in src/tickertape/lookup/table.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
def export_csv(self, filepath: Path | str) -> Path:
    """Export the full lookup table to a CSV file."""
    target = Path(filepath)
    target.parent.mkdir(parents=True, exist_ok=True)
    records = [m.model_dump() for m in self.all()]
    if not records:
        fieldnames = list(ISINMapping.model_fields.keys())
    else:
        fieldnames = list(records[0].keys())

    with open(target, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(records)

    return target

export_json 

export_json(filepath: Path | str) -> Path

Export the full lookup table to a JSON file.

Source code in src/tickertape/lookup/table.py 
325
326
327
328
329
330
331
332
def export_json(self, filepath: Path | str) -> Path:
    """Export the full lookup table to a JSON file."""
    target = Path(filepath)
    target.parent.mkdir(parents=True, exist_ok=True)
    records = [m.model_dump() for m in self.all()]
    with open(target, "w", encoding="utf-8") as f:
        json.dump(records, f, indent=2)
    return target

find_funds 

find_funds(
    *,
    sector: Optional[str] = None,
    subsector: Optional[str] = None,
    fund_type: Optional[str] = None,
    plan: Optional[str] = None,
    option: Optional[str] = None,
    risk_level: Optional[str] = None,
    benchmark: Optional[str] = None,
    amc: Optional[str] = None,
    limit: int = 100
) -> list[ISINMapping]

Gather mutual funds sharing similar classification attributes (peers).

Useful for AI agents to compare funds in the same category or benchmark.

Source code in src/tickertape/lookup/table.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def find_funds(
    self,
    *,
    sector: Optional[str] = None,
    subsector: Optional[str] = None,
    fund_type: Optional[str] = None,
    plan: Optional[str] = None,
    option: Optional[str] = None,
    risk_level: Optional[str] = None,
    benchmark: Optional[str] = None,
    amc: Optional[str] = None,
    limit: int = 100,
) -> list[ISINMapping]:
    """Gather mutual funds sharing similar classification attributes (peers).

    Useful for AI agents to compare funds in the same category or benchmark.
    """
    clauses: list[str] = []
    params: list[Any] = []

    if sector:
        clauses.append("LOWER(sector) = LOWER(?)")
        params.append(sector)
    if subsector:
        clauses.append("LOWER(subsector) = LOWER(?)")
        params.append(subsector)
    if fund_type:
        clauses.append("LOWER(fund_type) = LOWER(?)")
        params.append(fund_type)
    if plan:
        clauses.append("LOWER(plan) = LOWER(?)")
        params.append(plan)
    if option:
        clauses.append("LOWER(option) = LOWER(?)")
        params.append(option)
    if risk_level:
        clauses.append("LOWER(risk_level) = LOWER(?)")
        params.append(risk_level)
    if benchmark:
        clauses.append("LOWER(benchmark) = LOWER(?)")
        params.append(benchmark)
    if amc:
        clauses.append("LOWER(amc) LIKE LOWER(?)")
        params.append(f"%{amc}%")

    where_sql = f"WHERE {' AND '.join(clauses)}" if clauses else ""
    query = f"SELECT * FROM mf_isin_lookup {where_sql} ORDER BY name ASC LIMIT ?"
    params.append(limit)

    with self._get_connection() as conn:
        cursor = conn.execute(query, params)
        return [ISINMapping.model_validate(dict(row)) for row in cursor.fetchall()]

get 

get(isin: str) -> Optional[ISINMapping]

Fetch mapping for a specific ISIN.

Source code in src/tickertape/lookup/table.py 
120
121
122
123
124
125
126
127
128
129
130
def get(self, isin: str) -> Optional[ISINMapping]:
    """Fetch mapping for a specific ISIN."""
    clean_isin = isin.strip().upper()
    with self._get_connection() as conn:
        cursor = conn.execute(
            "SELECT * FROM mf_isin_lookup WHERE isin = ?", (clean_isin,)
        )
        row = cursor.fetchone()
        if row:
            return ISINMapping.model_validate(dict(row))
    return None

get_all_isins 

get_all_isins() -> set[str]

Return a set of all indexed ISINs.

Source code in src/tickertape/lookup/table.py 
259
260
261
262
263
def get_all_isins(self) -> set[str]:
    """Return a set of all indexed ISINs."""
    with self._get_connection() as conn:
        cursor = conn.execute("SELECT isin FROM mf_isin_lookup")
        return {row[0] for row in cursor.fetchall()}

get_all_record_ids 

get_all_record_ids() -> set[str]

Return a set of all indexed TickerTape record_ids.

Source code in src/tickertape/lookup/table.py 
253
254
255
256
257
def get_all_record_ids(self) -> set[str]:
    """Return a set of all indexed TickerTape record_ids."""
    with self._get_connection() as conn:
        cursor = conn.execute("SELECT record_id FROM mf_isin_lookup")
        return {row[0] for row in cursor.fetchall()}

get_batch 

get_batch(isins: list[str]) -> dict[str, ISINMapping]

Fetch mappings for multiple ISINs at once.

Source code in src/tickertape/lookup/table.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def get_batch(self, isins: list[str]) -> dict[str, ISINMapping]:
    """Fetch mappings for multiple ISINs at once."""
    clean_isins = [i.strip().upper() for i in isins if i.strip()]
    if not clean_isins:
        return {}

    placeholders = ",".join("?" for _ in clean_isins)
    result: dict[str, ISINMapping] = {}

    with self._get_connection() as conn:
        cursor = conn.execute(
            f"SELECT * FROM mf_isin_lookup WHERE isin IN ({placeholders})",
            clean_isins,
        )
        for row in cursor.fetchall():
            mapping = ISINMapping.model_validate(dict(row))
            result[mapping.isin] = mapping

    return result

get_by_record_id 

get_by_record_id(record_id: str) -> Optional[ISINMapping]

Fetch mapping by TickerTape record_id (e.g. 'M_QUNG').

Source code in src/tickertape/lookup/table.py 
152
153
154
155
156
157
158
159
160
161
162
def get_by_record_id(self, record_id: str) -> Optional[ISINMapping]:
    """Fetch mapping by TickerTape record_id (e.g. 'M_QUNG')."""
    clean_id = record_id.strip()
    with self._get_connection() as conn:
        cursor = conn.execute(
            "SELECT * FROM mf_isin_lookup WHERE record_id = ?", (clean_id,)
        )
        row = cursor.fetchone()
        if row:
            return ISINMapping.model_validate(dict(row))
    return None

get_peers 

get_peers(
    isin: str,
    match_plan: bool = True,
    match_option: bool = True,
    limit: int = 20,
) -> list[ISINMapping]

Get peer mutual funds in the same subsector/category as the given ISIN.

Parameters:

Name Type Description Default
isin str

Source fund ISIN.

 required
match_plan bool

If True, matches plan (e.g. Direct with Direct).

 True
match_option bool

If True, matches option (e.g. Growth with Growth).

 True
limit int

Maximum peers to return.

 20
Source code in src/tickertape/lookup/table.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def get_peers(
    self,
    isin: str,
    match_plan: bool = True,
    match_option: bool = True,
    limit: int = 20,
) -> list[ISINMapping]:
    """Get peer mutual funds in the same subsector/category as the given ISIN.

    Args:
        isin: Source fund ISIN.
        match_plan: If True, matches plan (e.g. Direct with Direct).
        match_option: If True, matches option (e.g. Growth with Growth).
        limit: Maximum peers to return.
    """
    fund = self.get(isin)
    if not fund or not fund.subsector:
        return []

    clauses = ["isin != ?", "LOWER(subsector) = LOWER(?)"]
    params: list[Any] = [fund.isin, fund.subsector]

    if match_plan and fund.plan:
        clauses.append("LOWER(plan) = LOWER(?)")
        params.append(fund.plan)
    if match_option and fund.option:
        clauses.append("LOWER(option) = LOWER(?)")
        params.append(fund.option)

    query = f"SELECT * FROM mf_isin_lookup WHERE {' AND '.join(clauses)} ORDER BY name ASC LIMIT ?"
    params.append(limit)

    with self._get_connection() as conn:
        cursor = conn.execute(query, params)
        return [ISINMapping.model_validate(dict(row)) for row in cursor.fetchall()]

upsert 

upsert(mapping: ISINMapping) -> None

Insert or update a single ISIN mapping.

Source code in src/tickertape/lookup/table.py 
265
266
267
def upsert(self, mapping: ISINMapping) -> None:
    """Insert or update a single ISIN mapping."""
    self.upsert_batch([mapping])

upsert_batch 

upsert_batch(mappings: list[ISINMapping]) -> None

Insert or update multiple ISIN mappings in a single atomic transaction.

Source code in src/tickertape/lookup/table.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def upsert_batch(self, mappings: list[ISINMapping]) -> None:
    """Insert or update multiple ISIN mappings in a single atomic transaction."""
    if not mappings:
        return

    query = """
    INSERT INTO mf_isin_lookup (
        isin, record_id, slug, name, amc, amc_code, sector, subsector,
        fund_type, fund_class, plan, option, risk_level, benchmark, url, nav, updated_at
    ) VALUES (
        :isin, :record_id, :slug, :name, :amc, :amc_code, :sector, :subsector,
        :fund_type, :fund_class, :plan, :option, :risk_level, :benchmark, :url, :nav, :updated_at
    )
    ON CONFLICT(isin) DO UPDATE SET
        record_id = excluded.record_id,
        slug = excluded.slug,
        name = excluded.name,
        amc = excluded.amc,
        amc_code = excluded.amc_code,
        sector = excluded.sector,
        subsector = excluded.subsector,
        fund_type = excluded.fund_type,
        fund_class = excluded.fund_class,
        plan = excluded.plan,
        option = excluded.option,
        risk_level = excluded.risk_level,
        benchmark = excluded.benchmark,
        url = excluded.url,
        nav = excluded.nav,
        updated_at = excluded.updated_at
    """
    payloads = [m.model_dump() for m in mappings]
    with self._get_connection() as conn:
        conn.executemany(query, payloads)
        conn.commit()

    logger.debug("Upserted %d mappings into SQLite lookup table", len(mappings))

Targeted ISIN Resolver

tickertape.lookup.resolver.ISINResolver

Resolves specific ISINs on demand using local sitemap slug matching and verification.

Source code in src/tickertape/lookup/resolver.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
class ISINResolver:
    """Resolves specific ISINs on demand using local sitemap slug matching and verification."""

    def __init__(
        self, client: TickerTapeClient, table: Optional[ISINLookupTable] = None
    ) -> None:
        self.client = client
        self.table = table or ISINLookupTable(cache_dir=client.cache_dir)

    async def resolve(
        self,
        isin: str,
        hint_name: Optional[str] = None,
        max_candidates: int = 3,
    ) -> Optional[ISINMapping]:
        """Resolve an ISIN to a TickerTape mapping.

        Checks the persistent SQLite table first. If not found, uses hint_name
        to search candidate slugs in the cached sitemap, verifies the page ISIN,
        and saves it to the SQLite lookup table upon match.

        Args:
            isin: Mutual fund ISIN string (e.g. 'INF966L01721').
            hint_name: Optional scheme name to narrow candidates.
            max_candidates: Maximum candidate pages to inspect.

        Returns:
            ISINMapping if successfully resolved, otherwise None.
        """
        clean_isin = isin.strip().upper()

        # 1. Check persistent SQLite table
        existing = self.table.get(clean_isin)
        if existing:
            return existing

        # 2. If no hint_name, we cannot narrow down 5,900+ sitemap URLs safely
        if not hint_name:
            logger.debug(
                "Cannot resolve ISIN %s without a hint name or pre-built index",
                clean_isin,
            )
            return None

        # 3. Load cached sitemap
        sitemap_items = await self.client.sitemap.get("mf")
        if not sitemap_items:
            return None

        # 4. Extract search tokens from hint_name
        tokens = [
            t.lower()
            for t in re.findall(r"[a-zA-Z0-9]+", hint_name)
            if t.lower() not in STOP_WORDS and len(t) > 2
        ]

        if not tokens:
            return None

        # 5. Score sitemap slugs based on token overlap
        scored_candidates = []
        for item in sitemap_items:
            slug_lower = item.url.lower()
            matched_count = sum(1 for token in tokens if token in slug_lower)
            if matched_count > 0:
                scored_candidates.append((matched_count, item))

        # Sort highest match count first
        scored_candidates.sort(key=lambda x: x[0], reverse=True)
        top_candidates = [item for _, item in scored_candidates[:max_candidates]]

        # 6. Fetch candidate pages and verify ISIN
        for candidate in top_candidates:
            try:
                fund = await self.client.mf.get(candidate.url)
                if fund.isin and fund.isin.strip().upper() == clean_isin:
                    raw_slug = fund.slug or candidate.url
                    clean_slug = raw_slug.split("/mutualfunds/")[-1].lstrip("/")

                    sec_info = fund.security_info
                    meta = fund.meta

                    mapping = ISINMapping(
                        isin=clean_isin,
                        record_id=fund.mf_id or candidate.record_id,
                        slug=clean_slug,
                        name=fund.name,
                        amc=(meta.amc if meta else None)
                        or (sec_info.amc if sec_info else None),
                        amc_code=sec_info.amc_code if sec_info else None,
                        sector=(sec_info.sector if sec_info else None)
                        or (meta.sector if meta else None),
                        subsector=(sec_info.subsector if sec_info else None)
                        or (meta.subsector if meta else None),
                        fund_type=meta.fund_type if meta else None,
                        fund_class=meta.type if meta else None,
                        plan=meta.plan if meta else None,
                        option=(sec_info.option if sec_info else None)
                        or (meta.option if meta else None),
                        risk_level=meta.risk_classification if meta else None,
                        benchmark=meta.benchmark_index if meta else None,
                        url=candidate.url,
                        nav=fund.nav,
                    )

                    self.table.upsert(mapping)
                    logger.info(
                        "Successfully resolved %s -> %s (%s)",
                        clean_isin,
                        mapping.record_id,
                        mapping.name,
                    )
                    return mapping
            except Exception as exc:
                logger.debug("Candidate check failed for %s: %s", candidate.url, exc)

        return None

resolve async 

resolve(
    isin: str,
    hint_name: Optional[str] = None,
    max_candidates: int = 3,
) -> Optional[ISINMapping]

Resolve an ISIN to a TickerTape mapping.

Checks the persistent SQLite table first. If not found, uses hint_name to search candidate slugs in the cached sitemap, verifies the page ISIN, and saves it to the SQLite lookup table upon match.

Parameters:

Name Type Description Default
isin str

Mutual fund ISIN string (e.g. 'INF966L01721').

 required
hint_name Optional[str]

Optional scheme name to narrow candidates.

 None
max_candidates int

Maximum candidate pages to inspect.

 3

Returns:

Type Description
Optional[ISINMapping]

ISINMapping if successfully resolved, otherwise None.
Source code in src/tickertape/lookup/resolver.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
async def resolve(
    self,
    isin: str,
    hint_name: Optional[str] = None,
    max_candidates: int = 3,
) -> Optional[ISINMapping]:
    """Resolve an ISIN to a TickerTape mapping.

    Checks the persistent SQLite table first. If not found, uses hint_name
    to search candidate slugs in the cached sitemap, verifies the page ISIN,
    and saves it to the SQLite lookup table upon match.

    Args:
        isin: Mutual fund ISIN string (e.g. 'INF966L01721').
        hint_name: Optional scheme name to narrow candidates.
        max_candidates: Maximum candidate pages to inspect.

    Returns:
        ISINMapping if successfully resolved, otherwise None.
    """
    clean_isin = isin.strip().upper()

    # 1. Check persistent SQLite table
    existing = self.table.get(clean_isin)
    if existing:
        return existing

    # 2. If no hint_name, we cannot narrow down 5,900+ sitemap URLs safely
    if not hint_name:
        logger.debug(
            "Cannot resolve ISIN %s without a hint name or pre-built index",
            clean_isin,
        )
        return None

    # 3. Load cached sitemap
    sitemap_items = await self.client.sitemap.get("mf")
    if not sitemap_items:
        return None

    # 4. Extract search tokens from hint_name
    tokens = [
        t.lower()
        for t in re.findall(r"[a-zA-Z0-9]+", hint_name)
        if t.lower() not in STOP_WORDS and len(t) > 2
    ]

    if not tokens:
        return None

    # 5. Score sitemap slugs based on token overlap
    scored_candidates = []
    for item in sitemap_items:
        slug_lower = item.url.lower()
        matched_count = sum(1 for token in tokens if token in slug_lower)
        if matched_count > 0:
            scored_candidates.append((matched_count, item))

    # Sort highest match count first
    scored_candidates.sort(key=lambda x: x[0], reverse=True)
    top_candidates = [item for _, item in scored_candidates[:max_candidates]]

    # 6. Fetch candidate pages and verify ISIN
    for candidate in top_candidates:
        try:
            fund = await self.client.mf.get(candidate.url)
            if fund.isin and fund.isin.strip().upper() == clean_isin:
                raw_slug = fund.slug or candidate.url
                clean_slug = raw_slug.split("/mutualfunds/")[-1].lstrip("/")

                sec_info = fund.security_info
                meta = fund.meta

                mapping = ISINMapping(
                    isin=clean_isin,
                    record_id=fund.mf_id or candidate.record_id,
                    slug=clean_slug,
                    name=fund.name,
                    amc=(meta.amc if meta else None)
                    or (sec_info.amc if sec_info else None),
                    amc_code=sec_info.amc_code if sec_info else None,
                    sector=(sec_info.sector if sec_info else None)
                    or (meta.sector if meta else None),
                    subsector=(sec_info.subsector if sec_info else None)
                    or (meta.subsector if meta else None),
                    fund_type=meta.fund_type if meta else None,
                    fund_class=meta.type if meta else None,
                    plan=meta.plan if meta else None,
                    option=(sec_info.option if sec_info else None)
                    or (meta.option if meta else None),
                    risk_level=meta.risk_classification if meta else None,
                    benchmark=meta.benchmark_index if meta else None,
                    url=candidate.url,
                    nav=fund.nav,
                )

                self.table.upsert(mapping)
                logger.info(
                    "Successfully resolved %s -> %s (%s)",
                    clean_isin,
                    mapping.record_id,
                    mapping.name,
                )
                return mapping
        except Exception as exc:
            logger.debug("Candidate check failed for %s: %s", candidate.url, exc)

    return None

Bulk ISIN Indexer

tickertape.lookup.indexer.ISINIndexer

Crawls mutual fund pages from sitemap and populates SQLite ISIN lookup table.

Source code in src/tickertape/lookup/indexer.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 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
 66
 67
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
class ISINIndexer:
    """Crawls mutual fund pages from sitemap and populates SQLite ISIN lookup table."""

    def __init__(
        self, client: TickerTapeClient, table: Optional[ISINLookupTable] = None
    ) -> None:
        self.client = client
        self.table = table or ISINLookupTable(cache_dir=client.cache_dir)

    async def build_index(
        self,
        *,
        force_refresh: bool = False,
        concurrency: int = 5,
        delay: float = 0.5,
        limit: Optional[int] = None,
        progress_callback: Optional[ProgressCallback] = None,
    ) -> int:
        """Build or update the SQLite ISIN lookup table from the mutual funds sitemap.

        Force Refresh Logic:
            When force_refresh=True:
            1. Fetches live sitemap XML from TickerTape.
            2. Parses sitemap and updates the local sitemap cache (.cache/tickertape/sitemaps/mf.json).
            3. Crawls fund pages and updates the SQLite database (.cache/tickertape/isin_lookup.db).

        When force_refresh=False:
            1. Loads sitemap from local disk cache.
            2. Skips record_ids already present in SQLite database (resumable).
            3. Crawls only missing funds and inserts into SQLite database.

        Args:
            force_refresh: If True, forces live sitemap fetch, updates sitemap cache, and re-indexes.
            concurrency: Number of concurrent HTTP workers.
            delay: Delay in seconds between requests per worker to respect rate limits.
            limit: Optional maximum number of funds to process (useful for testing or batch runs).
            progress_callback: Optional callback(completed, total, successful, failed).

        Returns:
            Number of mappings successfully indexed/updated in SQLite database.
        """
        # Step 1: Get sitemap items (force_refresh updates live XML and disk cache)
        if force_refresh:
            logger.info(
                "Force refresh triggered: fetching live sitemap and updating sitemap cache..."
            )
            sitemap_items = await self.client.sitemap.refresh("mf")
        else:
            sitemap_items = await self.client.sitemap.get("mf", force_refresh=False)

        if not sitemap_items:
            logger.warning("No sitemap items found to index")
            return 0

        # Step 2: Determine items to process
        if force_refresh:
            items_to_process = sitemap_items
        else:
            existing_ids = self.table.get_all_record_ids()
            items_to_process = [
                item for item in sitemap_items if item.record_id not in existing_ids
            ]

        if limit is not None:
            items_to_process = items_to_process[:limit]

        total = len(items_to_process)
        if total == 0:
            logger.info("All sitemap items are already indexed in SQLite lookup table.")
            return 0

        logger.info(
            "Starting ISIN indexing for %d funds (concurrency=%d)...",
            total,
            concurrency,
        )

        semaphore = asyncio.Semaphore(concurrency)
        completed = 0
        successful = 0
        failed = 0
        batch: list[ISINMapping] = []
        lock = asyncio.Lock()

        async def worker(item):
            nonlocal completed, successful, failed
            async with semaphore:
                try:
                    fund = await self.client.mf.get(item.url)
                    if fund.isin:
                        raw_slug = fund.slug or item.url
                        clean_slug = raw_slug.split("/mutualfunds/")[-1].lstrip("/")

                        sec_info = fund.security_info
                        meta = fund.meta

                        mapping = ISINMapping(
                            isin=fund.isin.strip().upper(),
                            record_id=fund.mf_id or item.record_id,
                            slug=clean_slug,
                            name=fund.name,
                            amc=(meta.amc if meta else None)
                            or (sec_info.amc if sec_info else None),
                            amc_code=sec_info.amc_code if sec_info else None,
                            sector=(sec_info.sector if sec_info else None)
                            or (meta.sector if meta else None),
                            subsector=(sec_info.subsector if sec_info else None)
                            or (meta.subsector if meta else None),
                            fund_type=meta.fund_type if meta else None,
                            fund_class=meta.type if meta else None,
                            plan=meta.plan if meta else None,
                            option=(sec_info.option if sec_info else None)
                            or (meta.option if meta else None),
                            risk_level=meta.risk_classification if meta else None,
                            benchmark=meta.benchmark_index if meta else None,
                            url=item.url,
                            nav=fund.nav,
                        )

                        async with lock:
                            batch.append(mapping)
                            successful += 1
                            if len(batch) >= 20:
                                self.table.upsert_batch(batch)
                                batch.clear()
                    else:
                        async with lock:
                            failed += 1
                except Exception as exc:
                    logger.debug("Failed to index %s: %s", item.url, exc)
                    async with lock:
                        failed += 1
                finally:
                    async with lock:
                        completed += 1
                        if progress_callback:
                            try:
                                progress_callback(completed, total, successful, failed)
                            except Exception:
                                pass

                if delay > 0:
                    await asyncio.sleep(delay)

        # Run workers concurrently
        tasks = [asyncio.create_task(worker(item)) for item in items_to_process]
        await asyncio.gather(*tasks)

        # Flush remaining batch
        if batch:
            self.table.upsert_batch(batch)
            batch.clear()

        logger.info(
            "Indexing completed: %d/%d successfully saved to SQLite (%d failed)",
            successful,
            total,
            failed,
        )
        return successful

build_index async 

build_index(
    *,
    force_refresh: bool = False,
    concurrency: int = 5,
    delay: float = 0.5,
    limit: Optional[int] = None,
    progress_callback: Optional[ProgressCallback] = None
) -> int

Build or update the SQLite ISIN lookup table from the mutual funds sitemap.

Force Refresh Logic

When force_refresh=True: 1. Fetches live sitemap XML from TickerTape. 2. Parses sitemap and updates the local sitemap cache (.cache/tickertape/sitemaps/mf.json). 3. Crawls fund pages and updates the SQLite database (.cache/tickertape/isin_lookup.db).

When force_refresh=False: 1. Loads sitemap from local disk cache. 2. Skips record_ids already present in SQLite database (resumable). 3. Crawls only missing funds and inserts into SQLite database.

Parameters:

Name Type Description Default
force_refresh bool

If True, forces live sitemap fetch, updates sitemap cache, and re-indexes.

 False
concurrency int

Number of concurrent HTTP workers.

 5
delay float

Delay in seconds between requests per worker to respect rate limits.

 0.5
limit Optional[int]

Optional maximum number of funds to process (useful for testing or batch runs).

 None
progress_callback Optional[ProgressCallback]

Optional callback(completed, total, successful, failed).

 None

Returns:

Type Description
int

Number of mappings successfully indexed/updated in SQLite database.
Source code in src/tickertape/lookup/indexer.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 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
 66
 67
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
async def build_index(
    self,
    *,
    force_refresh: bool = False,
    concurrency: int = 5,
    delay: float = 0.5,
    limit: Optional[int] = None,
    progress_callback: Optional[ProgressCallback] = None,
) -> int:
    """Build or update the SQLite ISIN lookup table from the mutual funds sitemap.

    Force Refresh Logic:
        When force_refresh=True:
        1. Fetches live sitemap XML from TickerTape.
        2. Parses sitemap and updates the local sitemap cache (.cache/tickertape/sitemaps/mf.json).
        3. Crawls fund pages and updates the SQLite database (.cache/tickertape/isin_lookup.db).

    When force_refresh=False:
        1. Loads sitemap from local disk cache.
        2. Skips record_ids already present in SQLite database (resumable).
        3. Crawls only missing funds and inserts into SQLite database.

    Args:
        force_refresh: If True, forces live sitemap fetch, updates sitemap cache, and re-indexes.
        concurrency: Number of concurrent HTTP workers.
        delay: Delay in seconds between requests per worker to respect rate limits.
        limit: Optional maximum number of funds to process (useful for testing or batch runs).
        progress_callback: Optional callback(completed, total, successful, failed).

    Returns:
        Number of mappings successfully indexed/updated in SQLite database.
    """
    # Step 1: Get sitemap items (force_refresh updates live XML and disk cache)
    if force_refresh:
        logger.info(
            "Force refresh triggered: fetching live sitemap and updating sitemap cache..."
        )
        sitemap_items = await self.client.sitemap.refresh("mf")
    else:
        sitemap_items = await self.client.sitemap.get("mf", force_refresh=False)

    if not sitemap_items:
        logger.warning("No sitemap items found to index")
        return 0

    # Step 2: Determine items to process
    if force_refresh:
        items_to_process = sitemap_items
    else:
        existing_ids = self.table.get_all_record_ids()
        items_to_process = [
            item for item in sitemap_items if item.record_id not in existing_ids
        ]

    if limit is not None:
        items_to_process = items_to_process[:limit]

    total = len(items_to_process)
    if total == 0:
        logger.info("All sitemap items are already indexed in SQLite lookup table.")
        return 0

    logger.info(
        "Starting ISIN indexing for %d funds (concurrency=%d)...",
        total,
        concurrency,
    )

    semaphore = asyncio.Semaphore(concurrency)
    completed = 0
    successful = 0
    failed = 0
    batch: list[ISINMapping] = []
    lock = asyncio.Lock()

    async def worker(item):
        nonlocal completed, successful, failed
        async with semaphore:
            try:
                fund = await self.client.mf.get(item.url)
                if fund.isin:
                    raw_slug = fund.slug or item.url
                    clean_slug = raw_slug.split("/mutualfunds/")[-1].lstrip("/")

                    sec_info = fund.security_info
                    meta = fund.meta

                    mapping = ISINMapping(
                        isin=fund.isin.strip().upper(),
                        record_id=fund.mf_id or item.record_id,
                        slug=clean_slug,
                        name=fund.name,
                        amc=(meta.amc if meta else None)
                        or (sec_info.amc if sec_info else None),
                        amc_code=sec_info.amc_code if sec_info else None,
                        sector=(sec_info.sector if sec_info else None)
                        or (meta.sector if meta else None),
                        subsector=(sec_info.subsector if sec_info else None)
                        or (meta.subsector if meta else None),
                        fund_type=meta.fund_type if meta else None,
                        fund_class=meta.type if meta else None,
                        plan=meta.plan if meta else None,
                        option=(sec_info.option if sec_info else None)
                        or (meta.option if meta else None),
                        risk_level=meta.risk_classification if meta else None,
                        benchmark=meta.benchmark_index if meta else None,
                        url=item.url,
                        nav=fund.nav,
                    )

                    async with lock:
                        batch.append(mapping)
                        successful += 1
                        if len(batch) >= 20:
                            self.table.upsert_batch(batch)
                            batch.clear()
                else:
                    async with lock:
                        failed += 1
            except Exception as exc:
                logger.debug("Failed to index %s: %s", item.url, exc)
                async with lock:
                    failed += 1
            finally:
                async with lock:
                    completed += 1
                    if progress_callback:
                        try:
                            progress_callback(completed, total, successful, failed)
                        except Exception:
                            pass

            if delay > 0:
                await asyncio.sleep(delay)

    # Run workers concurrently
    tasks = [asyncio.create_task(worker(item)) for item in items_to_process]
    await asyncio.gather(*tasks)

    # Flush remaining batch
    if batch:
        self.table.upsert_batch(batch)
        batch.clear()

    logger.info(
        "Indexing completed: %d/%d successfully saved to SQLite (%d failed)",
        successful,
        total,
        failed,
    )
    return successful

ISIN Mapping Model

tickertape.lookup.models.ISINMapping

Bases: BaseModel

Mapping entry linking a mutual fund ISIN to TickerTape record and categorization metadata.

Contains stable classification and grouping attributes used by AI agents to cluster and identify similar/peer mutual funds.

Source code in src/tickertape/lookup/models.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class ISINMapping(BaseModel):
    """Mapping entry linking a mutual fund ISIN to TickerTape record and categorization metadata.

    Contains stable classification and grouping attributes used by AI agents
    to cluster and identify similar/peer mutual funds.
    """

    model_config = ConfigDict(extra="ignore")

    isin: str
    record_id: str
    slug: str
    name: str
    amc: Optional[str] = None
    amc_code: Optional[str] = None
    sector: Optional[str] = (
        None  # Primary asset class (e.g. 'Equity', 'Debt', 'Hybrid')
    )
    subsector: Optional[str] = (
        None  # Granular category (e.g. 'Sectoral Fund - Infrastructure', 'Large Cap Fund')
    )
    fund_type: Optional[str] = None  # Fund type (e.g. 'Equity')
    fund_class: Optional[str] = None  # Structure type (e.g. 'Open', 'Close')
    plan: Optional[str] = None  # 'Direct' vs 'Regular'
    option: Optional[str] = None  # 'Growth' vs 'IDCW'
    risk_level: Optional[str] = (
        None  # Risk classification (e.g. 'Very High', 'Moderate', 'Low')
    )
    benchmark: Optional[str] = (
        None  # Benchmark Index (e.g. 'Nifty Infrastructure - TRI')
    )
    url: str
    nav: Optional[float] = None
    updated_at: str = Field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )