Resources Reference

Mutual Funds Resource

tickertape.resources.mf.MutualFundsResource

Bases: _BaseResource

Resource for fetching and parsing mutual fund details from TickerTape.

Source code in src/tickertape/resources/mf.py

class MutualFundsResource(_BaseResource):
    """Resource for fetching and parsing mutual fund details from TickerTape."""

    def __init__(self, client: Any) -> None:
        super().__init__(client)
        self._parser = MFParser()
        self.lookup = ISINLookupTable(cache_dir=client.cache_dir)
        self._resolver = ISINResolver(client=self._client, table=self.lookup)
        self._indexer = ISINIndexer(client=self._client, table=self.lookup)

    def _normalize_slug(self, slug_or_mfid: str) -> str:
        """Ensure slug path is formatted correctly."""
        slug = slug_or_mfid.strip()
        if slug.startswith(("http://", "https://")):
            return slug
        if slug.startswith("/mutualfunds/"):
            return slug
        if slug.startswith("/"):
            return f"/mutualfunds{slug}"
        return f"/mutualfunds/{slug}"

    async def get(self, slug_or_mfid: str) -> MutualFundDetail:
        """Fetch mutual fund details from TickerTape by slug or MFID.

        Args:
            slug_or_mfid: Slug (e.g. 'quant-infrastructure-fund-M_QUNG') or MFID ('M_QUNG').

        Returns:
            Parsed MutualFundDetail.
        """
        path = self._normalize_slug(slug_or_mfid)
        html_content = await self._client.request("GET", path)
        return self._parser.parse(html_content)

    async def get_by_isin(
        self,
        isin: str,
        hint_name: Optional[str] = None,
    ) -> Optional[MutualFundDetail]:
        """Fetch mutual fund details using an ISIN.

        Looks up the ISIN in the SQLite lookup table. If not found, attempts targeted
        resolution using hint_name against the sitemap and persists the result.

        Args:
            isin: Mutual fund ISIN (e.g. 'INF966L01721').
            hint_name: Optional scheme name to assist targeted resolution.

        Returns:
            Parsed MutualFundDetail if resolved, otherwise None.
        """
        mapping = await self._resolver.resolve(isin, hint_name=hint_name)
        if not mapping:
            return None
        return await self.get(mapping.slug)

    async def get_isin(self, slug_or_mfid: str) -> Optional[str]:
        """Convenience method to retrieve the ISIN for a mutual fund.

        Args:
            slug_or_mfid: Slug or MFID.

        Returns:
            ISIN string (e.g., 'INF966L01721') or None if not found.
        """
        fund = await self.get(slug_or_mfid)
        return fund.isin

    async def get_raw(self, slug_or_mfid: str) -> dict[str, Any]:
        """Fetch and extract the raw Next.js hydration payload dictionary."""
        path = self._normalize_slug(slug_or_mfid)
        html_content = await self._client.request("GET", path)
        return self._parser.extract_next_data(html_content)

    def get_peers(
        self,
        isin: str,
        match_plan: bool = True,
        match_option: bool = True,
        limit: int = 20,
    ) -> list[ISINMapping]:
        """Gather 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.
        """
        return self.lookup.get_peers(
            isin,
            match_plan=match_plan,
            match_option=match_option,
            limit=limit,
        )

    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]:
        """Find and gather mutual funds sharing similar classification attributes."""
        return self.lookup.find_funds(
            sector=sector,
            subsector=subsector,
            fund_type=fund_type,
            plan=plan,
            option=option,
            risk_level=risk_level,
            benchmark=benchmark,
            amc=amc,
            limit=limit,
        )

    async def build_isin_index(
        self,
        *,
        force_refresh: bool = False,
        concurrency: int = 5,
        delay: float = 0.5,
        limit: Optional[int] = None,
        progress_callback: Optional[Callable[[int, int, int, int], None]] = None,
    ) -> int:
        """Crawl sitemap URLs and populate the SQLite ISIN lookup table.

        When force_refresh=True:
            1. Refreshes live sitemap XML.
            2. Updates local sitemap disk cache (.cache/tickertape/sitemaps/mf.json).
            3. Crawls fund pages and updates SQLite database (.cache/tickertape/isin_lookup.db).

        When force_refresh=False:
            1. Uses cached sitemap.
            2. Resumes by crawling only unindexed funds.
            3. Updates SQLite database.

        Returns:
            Number of indexed mappings saved to SQLite.
        """
        return await self._indexer.build_index(
            force_refresh=force_refresh,
            concurrency=concurrency,
            delay=delay,
            limit=limit,
            progress_callback=progress_callback,
        )

    async def get_cached_by_isin(self, isin: str) -> ISINMapping | None:
        """
        Get table record by isin.

        Args:
            isin (str): Unique Identifier of scheme.

        Returns:
            ISINMapping: ISINMapping Object (Table Record)
        """
        mapping = await self._resolver.resolve(isin=isin)
        if not mapping:
            return None
        return mapping

    async def get_cached_by_isin_batch(
        self, isins: List[str]
    ) -> Dict[str, ISINMapping]:
        mappings = self.lookup.get_batch(isins)
        if not mappings:
            return {}
        return mappings

build_isin_index async

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

Crawl sitemap URLs and populate the SQLite ISIN lookup table.

When force_refresh=True: 1. Refreshes live sitemap XML. 2. Updates local sitemap disk cache (.cache/tickertape/sitemaps/mf.json). 3. Crawls fund pages and updates SQLite database (.cache/tickertape/isin_lookup.db).

When force_refresh=False: 1. Uses cached sitemap. 2. Resumes by crawling only unindexed funds. 3. Updates SQLite database.

Returns:

Type	Description
int	Number of indexed mappings saved to SQLite.

Source code in src/tickertape/resources/mf.py

async def build_isin_index(
    self,
    *,
    force_refresh: bool = False,
    concurrency: int = 5,
    delay: float = 0.5,
    limit: Optional[int] = None,
    progress_callback: Optional[Callable[[int, int, int, int], None]] = None,
) -> int:
    """Crawl sitemap URLs and populate the SQLite ISIN lookup table.

    When force_refresh=True:
        1. Refreshes live sitemap XML.
        2. Updates local sitemap disk cache (.cache/tickertape/sitemaps/mf.json).
        3. Crawls fund pages and updates SQLite database (.cache/tickertape/isin_lookup.db).

    When force_refresh=False:
        1. Uses cached sitemap.
        2. Resumes by crawling only unindexed funds.
        3. Updates SQLite database.

    Returns:
        Number of indexed mappings saved to SQLite.
    """
    return await self._indexer.build_index(
        force_refresh=force_refresh,
        concurrency=concurrency,
        delay=delay,
        limit=limit,
        progress_callback=progress_callback,
    )

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]

Find and gather mutual funds sharing similar classification attributes.

Source code in src/tickertape/resources/mf.py

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]:
    """Find and gather mutual funds sharing similar classification attributes."""
    return self.lookup.find_funds(
        sector=sector,
        subsector=subsector,
        fund_type=fund_type,
        plan=plan,
        option=option,
        risk_level=risk_level,
        benchmark=benchmark,
        amc=amc,
        limit=limit,
    )

get async

get(slug_or_mfid: str) -> MutualFundDetail

Fetch mutual fund details from TickerTape by slug or MFID.

Parameters:

Name	Type	Description	Default
slug_or_mfid	str	Slug (e.g. 'quant-infrastructure-fund-M_QUNG') or MFID ('M_QUNG').	required

Returns:

Type	Description
MutualFundDetail	Parsed MutualFundDetail.

Source code in src/tickertape/resources/mf.py

async def get(self, slug_or_mfid: str) -> MutualFundDetail:
    """Fetch mutual fund details from TickerTape by slug or MFID.

    Args:
        slug_or_mfid: Slug (e.g. 'quant-infrastructure-fund-M_QUNG') or MFID ('M_QUNG').

    Returns:
        Parsed MutualFundDetail.
    """
    path = self._normalize_slug(slug_or_mfid)
    html_content = await self._client.request("GET", path)
    return self._parser.parse(html_content)

get_by_isin async

get_by_isin(
    isin: str, hint_name: Optional[str] = None
) -> Optional[MutualFundDetail]

Fetch mutual fund details using an ISIN.

Looks up the ISIN in the SQLite lookup table. If not found, attempts targeted resolution using hint_name against the sitemap and persists the result.

Parameters:

Name	Type	Description	Default
isin	str	Mutual fund ISIN (e.g. 'INF966L01721').	required
hint_name	Optional[str]	Optional scheme name to assist targeted resolution.	None

Returns:

Type	Description
Optional[MutualFundDetail]	Parsed MutualFundDetail if resolved, otherwise None.

Source code in src/tickertape/resources/mf.py

async def get_by_isin(
    self,
    isin: str,
    hint_name: Optional[str] = None,
) -> Optional[MutualFundDetail]:
    """Fetch mutual fund details using an ISIN.

    Looks up the ISIN in the SQLite lookup table. If not found, attempts targeted
    resolution using hint_name against the sitemap and persists the result.

    Args:
        isin: Mutual fund ISIN (e.g. 'INF966L01721').
        hint_name: Optional scheme name to assist targeted resolution.

    Returns:
        Parsed MutualFundDetail if resolved, otherwise None.
    """
    mapping = await self._resolver.resolve(isin, hint_name=hint_name)
    if not mapping:
        return None
    return await self.get(mapping.slug)

get_cached_by_isin async

get_cached_by_isin(isin: str) -> ISINMapping | None

Get table record by isin.

Parameters:

Name	Type	Description	Default
isin	str	Unique Identifier of scheme.	required

Returns:

Name	Type	Description
ISINMapping	ISINMapping | None	ISINMapping Object (Table Record)

Source code in src/tickertape/resources/mf.py

async def get_cached_by_isin(self, isin: str) -> ISINMapping | None:
    """
    Get table record by isin.

    Args:
        isin (str): Unique Identifier of scheme.

    Returns:
        ISINMapping: ISINMapping Object (Table Record)
    """
    mapping = await self._resolver.resolve(isin=isin)
    if not mapping:
        return None
    return mapping

get_isin async

get_isin(slug_or_mfid: str) -> Optional[str]

Convenience method to retrieve the ISIN for a mutual fund.

Parameters:

Name	Type	Description	Default
slug_or_mfid	str	Slug or MFID.	required

Returns:

Type	Description
Optional[str]	ISIN string (e.g., 'INF966L01721') or None if not found.

Source code in src/tickertape/resources/mf.py

async def get_isin(self, slug_or_mfid: str) -> Optional[str]:
    """Convenience method to retrieve the ISIN for a mutual fund.

    Args:
        slug_or_mfid: Slug or MFID.

    Returns:
        ISIN string (e.g., 'INF966L01721') or None if not found.
    """
    fund = await self.get(slug_or_mfid)
    return fund.isin

get_peers

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

Gather 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/resources/mf.py

def get_peers(
    self,
    isin: str,
    match_plan: bool = True,
    match_option: bool = True,
    limit: int = 20,
) -> list[ISINMapping]:
    """Gather 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.
    """
    return self.lookup.get_peers(
        isin,
        match_plan=match_plan,
        match_option=match_option,
        limit=limit,
    )

get_raw async

get_raw(slug_or_mfid: str) -> dict[str, Any]

Fetch and extract the raw Next.js hydration payload dictionary.

Source code in src/tickertape/resources/mf.py

async def get_raw(self, slug_or_mfid: str) -> dict[str, Any]:
    """Fetch and extract the raw Next.js hydration payload dictionary."""
    path = self._normalize_slug(slug_or_mfid)
    html_content = await self._client.request("GET", path)
    return self._parser.extract_next_data(html_content)

---

Sitemap Resource

tickertape.resources.sitemap.SitemapResource

Bases: _BaseResource

Resource managing XML sitemaps with local persistent caching.

Source code in src/tickertape/resources/sitemap.py

class SitemapResource(_BaseResource):
    """Resource managing XML sitemaps with local persistent caching."""

    async def get(
        self,
        category: str = "mf",
        *,
        force_refresh: bool = False,
    ) -> list[SitemapURL]:
        """Get parsed sitemap URLs for a category.

        By default, loads from persistent disk cache (.cache/tickertape)
        unless force_refresh is True or cache does not exist.

        Args:
            category: Category key ('mf', 'stocks', 'etf', etc.) or custom sitemap key.
            force_refresh: If True, bypasses local cache, fetches live XML, and updates cache.

        Returns:
            List of parsed SitemapURL items.
        """
        category_key = category.lower().strip()

        # Check disk cache unless forced refresh
        if not force_refresh:
            cached_items = self._client.cache_manager.load(category_key)
            if cached_items is not None:
                logger.debug(
                    "Loaded %d sitemap items for '%s' from cache",
                    len(cached_items),
                    category_key,
                )
                return cached_items

        # Fetch live XML and refresh cache
        return await self.refresh(category_key)

    async def refresh(self, category: str = "mf") -> list[SitemapURL]:
        """Fetch live XML sitemap, parse it, update the local cache, and return items.

        Args:
            category: Category key ('mf', 'stocks', 'etf', etc.) or full URL.

        Returns:
            List of refreshed SitemapURL items.
        """
        category_key = category.lower().strip()

        if category_key.startswith(("http://", "https://")):
            url = category
            # Derive cache category key from URL or keep generic
            cache_category = category.rstrip("/").split("/")[-2] or "custom"
        else:
            url = self._client.sitemap_urls.get(category_key)
            if not url:
                raise ValueError(
                    f"Unknown sitemap category '{category}'. Available categories: {list(self._client.sitemap_urls.keys())}"
                )
            cache_category = category_key

        logger.info("Fetching live sitemap XML from %s", url)
        response_text = await self._client.request("GET", url)

        parser = SitemapParser(response_text)
        parsed_result = parser.parse()

        if not isinstance(parsed_result, list):
            raise TickerTapeParseError(
                f"Unexpected parsed sitemap result type: {type(parsed_result)}"
            )

        if parsed_result and isinstance(parsed_result[0], SitemapReference):
            raise TickerTapeParseError(
                f"Sitemap at {url} is a sitemap index containing child sitemaps. Use get_references() instead."
            )

        items: list[SitemapURL] = parsed_result  # type: ignore

        # Persist to disk cache
        self._client.cache_manager.save(cache_category, items)
        return items

    async def get_references(self, url_or_category: str) -> list[SitemapReference]:
        """Fetch and parse a sitemap index XML into a list of SitemapReference items."""
        if url_or_category.startswith(("http://", "https://")):
            url = url_or_category
        else:
            url = self._client.sitemap_urls.get(url_or_category.lower().strip(), "")
            if not url:
                raise ValueError(f"Unknown sitemap category: {url_or_category}")

        response_text = await self._client.request("GET", url)
        parser = SitemapParser(response_text)
        parsed = parser.parse()

        return [item for item in parsed if isinstance(item, SitemapReference)]

    def is_cached(self, category: str = "mf") -> bool:
        """Check if local cache exists for a given category."""
        return self._client.cache_manager.exists(category.lower().strip())

    def clear_cache(self, category: Optional[str] = None) -> None:
        """Clear the cached sitemap file for category or all sitemaps."""
        self._client.cache_manager.clear(category)

clear_cache

clear_cache(category: Optional[str] = None) -> None

Clear the cached sitemap file for category or all sitemaps.

Source code in src/tickertape/resources/sitemap.py

def clear_cache(self, category: Optional[str] = None) -> None:
    """Clear the cached sitemap file for category or all sitemaps."""
    self._client.cache_manager.clear(category)

get async

get(
    category: str = "mf", *, force_refresh: bool = False
) -> list[SitemapURL]

Get parsed sitemap URLs for a category.

By default, loads from persistent disk cache (.cache/tickertape) unless force_refresh is True or cache does not exist.

Parameters:

Name	Type	Description	Default
category	str	Category key ('mf', 'stocks', 'etf', etc.) or custom sitemap key.	'mf'
force_refresh	bool	If True, bypasses local cache, fetches live XML, and updates cache.	False

Returns:

Type	Description
list[SitemapURL]	List of parsed SitemapURL items.

Source code in src/tickertape/resources/sitemap.py

async def get(
    self,
    category: str = "mf",
    *,
    force_refresh: bool = False,
) -> list[SitemapURL]:
    """Get parsed sitemap URLs for a category.

    By default, loads from persistent disk cache (.cache/tickertape)
    unless force_refresh is True or cache does not exist.

    Args:
        category: Category key ('mf', 'stocks', 'etf', etc.) or custom sitemap key.
        force_refresh: If True, bypasses local cache, fetches live XML, and updates cache.

    Returns:
        List of parsed SitemapURL items.
    """
    category_key = category.lower().strip()

    # Check disk cache unless forced refresh
    if not force_refresh:
        cached_items = self._client.cache_manager.load(category_key)
        if cached_items is not None:
            logger.debug(
                "Loaded %d sitemap items for '%s' from cache",
                len(cached_items),
                category_key,
            )
            return cached_items

    # Fetch live XML and refresh cache
    return await self.refresh(category_key)

get_references async

get_references(
    url_or_category: str,
) -> list[SitemapReference]

Fetch and parse a sitemap index XML into a list of SitemapReference items.

Source code in src/tickertape/resources/sitemap.py

async def get_references(self, url_or_category: str) -> list[SitemapReference]:
    """Fetch and parse a sitemap index XML into a list of SitemapReference items."""
    if url_or_category.startswith(("http://", "https://")):
        url = url_or_category
    else:
        url = self._client.sitemap_urls.get(url_or_category.lower().strip(), "")
        if not url:
            raise ValueError(f"Unknown sitemap category: {url_or_category}")

    response_text = await self._client.request("GET", url)
    parser = SitemapParser(response_text)
    parsed = parser.parse()

    return [item for item in parsed if isinstance(item, SitemapReference)]

is_cached

is_cached(category: str = 'mf') -> bool

Check if local cache exists for a given category.

Source code in src/tickertape/resources/sitemap.py

def is_cached(self, category: str = "mf") -> bool:
    """Check if local cache exists for a given category."""
    return self._client.cache_manager.exists(category.lower().strip())

refresh async

refresh(category: str = 'mf') -> list[SitemapURL]

Fetch live XML sitemap, parse it, update the local cache, and return items.

Parameters:

Name	Type	Description	Default
category	str	Category key ('mf', 'stocks', 'etf', etc.) or full URL.	'mf'

Returns:

Type	Description
list[SitemapURL]	List of refreshed SitemapURL items.

Source code in src/tickertape/resources/sitemap.py

async def refresh(self, category: str = "mf") -> list[SitemapURL]:
    """Fetch live XML sitemap, parse it, update the local cache, and return items.

    Args:
        category: Category key ('mf', 'stocks', 'etf', etc.) or full URL.

    Returns:
        List of refreshed SitemapURL items.
    """
    category_key = category.lower().strip()

    if category_key.startswith(("http://", "https://")):
        url = category
        # Derive cache category key from URL or keep generic
        cache_category = category.rstrip("/").split("/")[-2] or "custom"
    else:
        url = self._client.sitemap_urls.get(category_key)
        if not url:
            raise ValueError(
                f"Unknown sitemap category '{category}'. Available categories: {list(self._client.sitemap_urls.keys())}"
            )
        cache_category = category_key

    logger.info("Fetching live sitemap XML from %s", url)
    response_text = await self._client.request("GET", url)

    parser = SitemapParser(response_text)
    parsed_result = parser.parse()

    if not isinstance(parsed_result, list):
        raise TickerTapeParseError(
            f"Unexpected parsed sitemap result type: {type(parsed_result)}"
        )

    if parsed_result and isinstance(parsed_result[0], SitemapReference):
        raise TickerTapeParseError(
            f"Sitemap at {url} is a sitemap index containing child sitemaps. Use get_references() instead."
        )

    items: list[SitemapURL] = parsed_result  # type: ignore

    # Persist to disk cache
    self._client.cache_manager.save(cache_category, items)
    return items

---

Base Resource

tickertape.resources.base._BaseResource

Base resource associated with a TickerTapeClient instance.

Source code in src/tickertape/resources/base.py

class _BaseResource:
    """Base resource associated with a TickerTapeClient instance."""

    def __init__(self, client: TickerTapeClient) -> None:
        self._client = client