### Install VNStock Library

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Instructions for installing the vnstock Python package using pip and verifying the installation by checking the version. It also mentions the separate installation process for sponsored packages like vnstock_data.

```bash
# Install vnstock from PyPI
pip install vnstock

# Verify installation
python -c "import vnstock; print(vnstock.__version__)"

# For sponsored packages (vnstock_data)
# Follow the installation guide at: https://vnstocks.com/onboard-member
```

--------------------------------

### Work with Different Timeframes for Stock Data (Python)

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Demonstrates how to fetch historical stock data using various timeframes (resolutions) supported by the vnstock library. It shows examples for minute, hour, daily, weekly, and monthly data. Dependencies include 'vnstock.Quote' and 'vnstock.core.types.TimeFrame'. Outputs Pandas DataFrames for each requested timeframe.

```python
from vnstock import Quote
from vnstock.core.types import TimeFrame

quote = Quote(source="vci", symbol="VCI")

# VCI supports all 21 timeframe formats:
# Standard (9): "1m", "5m", "15m", "30m", "1H", "4h", "1D", "1W", "1M"
# Lowercase (3): "1h", "1d", "1w"
# Single char aliases (5): "m", "h", "d", "w", "M"
# Full names (5): "minute", "hour", "day", "week", "month"

# 1-minute data (intraday scalping)
df_1m = quote.history(
    start="2024-12-02",
    end="2024-12-03",
    resolution=TimeFrame.MINUTE_1  # or "1m", "m", "minute"
)
print(f"1-minute bars: {len(df_1m)}")

# 1-hour data (day trading)
df_1h = quote.history(
    start="2024-12-02",
    end="2024-12-03",
    resolution="1H"  # or "1h", "h", "hour", TimeFrame.HOUR_1
)
print(f"1-hour bars: {len(df_1h)}")

# Daily data (swing trading)
df_daily = quote.history(
    start="2024-01-01",
    end="2024-12-31",
    resolution="1D"  # or "1d", "d", "day", "DAILY", TimeFrame.DAILY
)
print(f"Daily bars: {len(df_daily)}")

# Weekly data (position trading)
df_weekly = quote.history(
    start="2023-01-01",
    end="2024-12-31",
    resolution=TimeFrame.WEEKLY  # or "1W", "1w", "w", "week"
)
print(f"Weekly bars: {len(df_weekly)}")

# Monthly data (long-term investing)
df_monthly = quote.history(
    start="2020-01-01",
    end="2024-12-31",
    resolution=TimeFrame.MONTHLY  # or "1M", "M", "month"
)
print(f"Monthly bars: {len(df_monthly)}")

# Note: Not all providers support all timeframes
# VCI: Full support (all 21 formats)
# TCBS: Daily, Weekly, Monthly only
# MSN: Daily only
```

--------------------------------

### Unit Testing Custom Pipeline Components (Python)

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/03-custom-pipelines.md

This Python script provides unit tests for the custom pipeline components (Fetcher, Validator, Transformer). It uses the `unittest` framework and `pandas` for data manipulation. The `setUp` method creates sample data, and individual test methods verify the functionality of each component, checking for expected outputs and data integrity.

```python
import unittest
import pandas as pd
import numpy as np

class TestCustomPipeline(unittest.TestCase):
    
    def setUp(self):
        # Create sample data
        self.valid_data = pd.DataFrame({
            'time': pd.date_range('2024-01-01', periods=50),
            'open': np.random.uniform(60, 65, 50),
            'high': np.random.uniform(63, 67, 50),
            'low': np.random.uniform(58, 62, 50),
            'close': np.random.uniform(60, 65, 50),
            'volume': np.random.randint(1000000, 10000000, 50)
        })
        
        # Ensure OHLC logic
        self.valid_data['low'] = self.valid_data[['open', 'high', 'low', 'close']].min(axis=1)
        self.valid_data['high'] = self.valid_data[['open', 'high', 'low', 'close']].max(axis=1)
    
    def test_fetcher(self):
        fetcher = ProductionFetcher()
        df = fetcher._vn_call("VCB", start="2024-01-01", end="2024-12-02")
        self.assertGreater(len(df), 0)
        self.assertIn('close', df.columns)
    
    def test_validator(self):
        validator = ProductionValidator()
        self.assertTrue(validator.validate(self.valid_data))
    
    def test_transformer(self):
        transformer = ProductionTransformer()
        result = transformer.transform(self.valid_data)
        self.assertIn('sma20', result.columns)
        self.assertIn('rsi', result.columns)
        self.assertIn('macd', result.columns)

if __name__ == '__main__':
    unittest.main()
```

--------------------------------

### Build and Run Docker Pipeline

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/05-best-practices.md

Commands to build the Docker image for the vnstock pipeline and run it, mounting a local data directory into the container. This setup is crucial for executing the data processing pipeline.

```bash
docker build -t vnstock-pipeline:latest .
docker run -v $(pwd)/data:/app/data vnstock-pipeline:latest
```

--------------------------------

### Production Pipeline Example: Fetch, Validate, and Transform VN100 Stocks

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/03-custom-pipelines.md

This Python script outlines a complete production pipeline for fetching, validating, and transforming VN100 stock data. It defines custom `ProductionFetcher` and `ProductionValidator` classes inheriting from base classes. The fetcher uses `vnstock_data.Quote` to retrieve historical data, while the validator implements specific checks for data completeness and integrity.

```python
"""
Production pipeline: Fetch VN100 stocks, enrich with indicators, export
"""

from vnstock_pipeline.core.scheduler import Scheduler
from vnstock_pipeline.template.vnstock import VNFetcher, VNValidator, VNTransformer
from vnstock_pipeline.core.exporter import Exporter
from vnstock_data import Quote, Finance
import pandas as pd
import numpy as np
import logging
from datetime import datetime

# Setup
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ============= FETCHER =============
class ProductionFetcher(VNFetcher):
    def _vn_call(self, ticker: str, **kwargs) -> pd.DataFrame:
        try:
            quote = Quote(source="vci", symbol=ticker)
            df = quote.history(
                start=kwargs.get("start", "2024-01-01"),
                end=kwargs.get("end", "2024-12-02"),
                interval=kwargs.get("interval", "1D")
            )
            logger.info(f"✅ Fetched {len(df)} rows for {ticker}")
            return df
        except Exception as e:
            logger.error(f"❌ Fetch failed for {ticker}: {e}")
            return pd.DataFrame()

# ============= VALIDATOR =============
class ProductionValidator(VNValidator):
    required_columns = ["time", "open", "high", "low", "close", "volume"]
    
    def validate(self, data: pd.DataFrame) -> bool:
        if not super().validate(data):
            return False
        
        # Min rows
        if len(data) < 20:
            return False
        
        # OHLC checks
        if (data['high'] < data['low']).any():
            return False
        
        # Volume checks
        if (data['volume'] <= 0).any():
            return False
        
        return True

```

--------------------------------

### Docker Deployment Configuration (Dockerfile)

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/05-best-practices.md

This Dockerfile sets up a production environment for the vnstock-agent. It uses a slim Python 3.10 image, installs project dependencies, copies the application code, and creates a directory for data storage.

```dockerfile
FROM python:3.10-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy code
COPY vnstock_pipeline/ ./vnstock_pipeline/
COPY pipeline_script.py .

# Create data directory
RUN mkdir -p /app/data

```

--------------------------------

### Graceful WebSocket Resource Management

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/04-streaming.md

This Python snippet demonstrates best practices for managing WebSocket resources. It shows how to use a context manager (`with websocket.WebSocketApp(...) as ws:`) for automatic setup and teardown, ensuring the WebSocket connection is properly handled. It also includes an `on_close` function for performing cleanup actions when the connection is closed.

```python
# Use context managers
with websocket.WebSocketApp(url) as ws:
    ws.run_forever()

# Close gracefully
def on_close(ws, code, msg):
    print("Closed")
    ws.close()
```

--------------------------------

### Running the Stock Data Processing Pipeline (Python)

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/03-custom-pipelines.md

This script demonstrates how to set up and run a data processing pipeline. It fetches VN100 tickers, creates a scheduler with Fetcher, Validator, Transformer, and Exporter components, and then executes the pipeline for the specified tickers with date ranges. It requires vnstock, pandas, and custom scheduler/component classes.

```python
if __name__ == "__main__":
    # Get VN100 tickers
    from vnstock import Vnstock
    stock = Vnstock().stock(symbol="VCB", source="VCI")
    tickers = stock.listing.symbols_by_group("VN100").tolist()
    
    # Create scheduler
    scheduler = Scheduler(
        ProductionFetcher(),
        ProductionValidator(),
        ProductionTransformer(),
        ProductionExporter("./vn100_enriched"),
        retry_attempts=3,
        backoff_factor=2.0
    )
    
    # Run
    logger.info(f"Starting pipeline for {len(tickers)} tickers")
    scheduler.run(
        tickers,
        fetcher_kwargs={
            "start": "2024-01-01",
            "end": "2024-12-02",
            "interval": "1D"
        }
    )
    
    logger.info("✅ Pipeline completed!")
```

--------------------------------

### Historical and Real-time Price Data with VNStock Quote API

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Shows how to retrieve historical and real-time price data using the Quote adapter from vnstock. It supports multiple timeframes (daily, intraday, weekly) and provides examples for calculating technical indicators like SMA, daily return, and cumulative return. It also demonstrates fetching intraday tick data and price depth (order book).

```python
from vnstock import Quote
from vnstock.core.types import TimeFrame
import datetime

# Initialize Quote adapter
quote = Quote(source="vci", symbol="VCI", show_log=False)

# Get daily historical data (supports 21 timeframe formats)
df_daily = quote.history(
    start="2024-01-01",
    end="2024-12-31",
    resolution=TimeFrame.DAILY  # or "1D", "1d", "d", "day"
)
print(f"Daily data: {len(df_daily)} rows")
print(df_daily.head())
# Columns: time, open, high, low, close, volume

# Get intraday 5-minute data
df_5m = quote.history(
    start="2024-12-02",
    end="2024-12-03",
    resolution="5m"  # or TimeFrame.MINUTE_5
)
print(f"5-minute data: {len(df_5m)} rows")

# Get weekly data for trend analysis
df_weekly = quote.history(
    start="2023-01-01",
    end="2024-12-31",
    resolution=TimeFrame.WEEKLY  # or "1W", "w", "week"
)

# Calculate technical indicators
df_daily['SMA_20'] = df_daily['close'].rolling(window=20).mean()
df_daily['SMA_50'] = df_daily['close'].rolling(window=50).mean()
df_daily['daily_return'] = df_daily['close'].pct_change()
df_daily['cumulative_return'] = (1 + df_daily['daily_return']).cumprod() - 1

print(df_daily[['close', 'SMA_20', 'SMA_50', 'daily_return']].tail())

# Get intraday tick data
intraday_df = quote.intraday(page_size=100)
print(f"Intraday ticks: {len(intraday_df)}")

# Get price depth (order book)
depth = quote.price_depth()
if depth:
    best_bid = depth['bid'][0]
    best_ask = depth['ask'][0]
    print(f"Best Bid: {best_bid['price']} x {best_bid['volume']}")
    print(f"Best Ask: {best_ask['price']} x {best_ask['volume']}")
```

--------------------------------

### Corporate Information with VNStock Company API

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Illustrates how to access corporate information such as company profiles and major shareholders using the Company adapter from vnstock. It shows how to initialize the adapter with a specific source and symbol, and retrieve overview data and shareholder details.

```python
from vnstock import Company

# Initialize Company adapter
company = Company(source="vci", symbol="VCI")

# Get company overview
overview = company.overview()
print(overview.columns.tolist())
print(overview.head())

# Get major shareholders
shareholders = company.shareholders()
print(f"Total shareholders: {len(shareholders)}")
print(shareholders[['share_holder', 'quantity', 'share_own_percent']].head())
```

--------------------------------

### Stock Discovery and Filtering with VNStock Listing API

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Demonstrates how to use the Listing adapter from the vnstock library to discover and filter stocks. It covers fetching all stock symbols, filtering by exchange (HOSE, HNX, UPCOM), retrieving stocks by index group (VN30, VNMID), and accessing industry classifications.

```python
from vnstock import Listing

# Initialize Listing adapter with VCI source
listing = Listing(source="vci", random_agent=False, show_log=True)

# Get all stock symbols as DataFrame
all_stocks = listing.all_symbols(to_df=True)
print(f"Total stocks: {len(all_stocks)}")
print(all_stocks.head())
# Output columns: symbol, company_name, exchange, industry

# Filter by exchange (HOSE, HNX, UPCOM)
hose_symbols = listing.symbols_by_exchange(exchange="HOSE")
print(f"HOSE stocks: {len(hose_symbols)}")

# Get stocks by index group (VN30, VNMID, VNSML, VNIT, VNFIN, etc.)
vn30_stocks = listing.symbols_by_group(group="VN30")
print(f"VN30 members: {vn30_stocks}")

# Get industry classification
industries_df = listing.symbols_by_industries(to_df=True)
finance_stocks = industries_df[industries_df['industry_name'] == 'Finance']
print(finance_stocks[['symbol', 'company_name']].head())

# Get ICB classification details
icb_df = listing.industries_icb()
print(icb_df[['icb_name', 'super_group']].head())
```

--------------------------------

### Production Data Transformation with Technical Indicators (Python)

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/03-custom-pipelines.md

This class extends VNTransformer to enrich stock data with common technical indicators such as Simple Moving Averages (SMA), Relative Strength Index (RSI), Moving Average Convergence Divergence (MACD), volatility, and price change percentage. It takes a pandas DataFrame as input and returns the DataFrame with added indicator columns. Ensure pandas is installed.

```python
class ProductionTransformer(VNTransformer):
    def transform(self, data: pd.DataFrame) -> pd.DataFrame:
        df = super().transform(data)
        
        # Moving Averages
        df['sma20'] = df['close'].rolling(20).mean()
        df['sma50'] = df['close'].rolling(50).mean()
        
        # RSI
        delta = df['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
        rs = gain / loss
        df['rsi'] = 100 - (100 / (1 + rs))
        
        # MACD
        exp1 = df['close'].ewm(span=12, adjust=False).mean()
        exp2 = df['close'].ewm(span=26, adjust=False).mean()
        df['macd'] = exp1 - exp2
        df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
        
        # Volatility
        df['volatility_30d'] = df['close'].pct_change().rolling(30).std() * 100
        
        # Price change
        df['change_pct'] = df['close'].pct_change() * 100
        
        logger.info(f"Enriched {len(df)} rows with indicators")
        return df
```

--------------------------------

### Multi-Company Analysis and Comparison with vnstock

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

This code snippet demonstrates how to perform multi-company analysis by comparing financial metrics and stock performance across multiple companies. It fetches VN30 stock symbols, collects historical price data, calculates annualized returns and volatility, computes Sharpe Ratios, and compares the number of balance sheet and income statement records for selected companies. It uses Listing, Quote, and Finance adapters.

```python
from vnstock import Listing, Quote, Finance
from vnstock.core.types import TimeFrame
import pandas as pd

# Get VN30 stocks using 'vci' as the source
listing = Listing(source="vci")
vn30_stocks = listing.symbols_by_group(group="VN30")

# Collect price data for multiple symbols
price_data = {}
# Limit to first 5 stocks for demonstration purposes
for symbol in vn30_stocks[:5]:
    try:
        # Initialize Quote adapter for each symbol
        quote = Quote(source="vci", symbol=symbol)
        # Fetch daily historical data for the year 2024
        df = quote.history(
            start="2024-01-01",
            end="2024-12-31",
            resolution=TimeFrame.DAILY
        )
        # Store closing prices if data is not empty
        if not df.empty:
            price_data[symbol] = df['close']
    except Exception as e:
        print(f"Error fetching price data for {symbol}: {e}")

# Create a comparison DataFrame from the collected price data
comparison_df = pd.DataFrame(price_data)
print("Comparison DataFrame (Close Prices):")
print(comparison_df.head())

# Calculate annualized returns and volatility
# Assumes 252 trading days in a year
returns = comparison_df.pct_change().mean() * 252
volatility = comparison_df.pct_change().std() * (252 ** 0.5)

# Create a performance DataFrame including Sharpe Ratio
performance = pd.DataFrame({
    'Annual Return': returns,
    'Annual Volatility': volatility,
    'Sharpe Ratio': returns / volatility
})
print("\nPerformance Metrics (Sorted by Sharpe Ratio):")
print(performance.sort_values('Sharpe Ratio', ascending=False))

# Compare financial metrics across companies
financial_comparison = []
# Iterate through the same first 5 stocks for financial data comparison
for symbol in vn30_stocks[:5]:
    try:
        # Initialize Finance adapter for each symbol
        finance = Finance(source="vci", symbol=symbol)
        # Fetch yearly balance sheet and income statement
        balance_sheet = finance.balance_sheet(period="year")
        income_statement = finance.income_statement(period="year")

        # Append data if both statements are available
        if not balance_sheet.empty and not income_statement.empty:
            financial_comparison.append({
                'symbol': symbol,
                'bs_records': len(balance_sheet),
                'is_records': len(income_statement)
            })
    except Exception as e:
        print(f"Error fetching financial data for {symbol}: {e}")

# Create a DataFrame for financial comparison results
comparison_table = pd.DataFrame(financial_comparison)
print("\nFinancial Metrics Comparison:")
print(comparison_table)

```

--------------------------------

### Advanced Stock Screening and Filtering with vnstock

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

This snippet illustrates advanced stock screening by filtering stocks based on multiple criteria. It retrieves all HOSE stocks along with their industry information, merges this data, and prepares it for further filtering. This serves as a foundation for building custom stock screening tools using the Listing adapter.

```python
from vnstock import Listing, Quote, Finance
from vnstock.core.types import TimeFrame
import pandas as pd

# Initialize Listing adapter using 'vci' as the source
listing = Listing(source="vci")

# Get all HOSE stocks with industry information, returned as pandas DataFrames
all_stocks = listing.all_symbols(to_df=True)
industries = listing.symbols_by_industries(to_df=True)

# Merge the two DataFrames on the 'symbol' column using a left merge
# This ensures all stocks are kept, and industry info is added where available
merged = all_stocks.merge(industries, on='symbol', how='left')

# The 'merged' DataFrame now contains stock symbols and their corresponding industries,
# ready for further filtering and analysis.
print("Merged stock data with industry information (first 5 rows):")
print(merged.head())

# Example of further filtering (e.g., find stocks in the 'Banking' industry):
# banking_stocks = merged[merged['industry'] == 'Ngân hàng'] # Adjust 'Ngân hàng' if industry names differ
# print("\nBanking stocks:")
# print(banking_stocks)

```

--------------------------------

### Retrieve Financial Statements and Ratios with vnstock Finance API

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

This snippet shows how to use the vnstock Finance API to retrieve yearly balance sheets, quarterly income statements, and yearly cash flow statements for a specific company. It also demonstrates fetching financial ratios and exporting the data to an Excel file. The Finance adapter requires a source, symbol, and an option to show logs.

```python
from vnstock import Finance
import pandas as pd

# Initialize Finance adapter
# Ensure 'source' and 'symbol' are valid for your vnstock setup
finance = Finance(source="vci", symbol="VCI", show_log=False)

# Get balance sheet (yearly)
balance_sheet = finance.balance_sheet(period="year")
print(f"Balance sheet records: {len(balance_sheet)}")
# Display first 10 columns for brevity
print(balance_sheet.columns.tolist()[:10])

# Get income statement (quarterly)
income_statement = finance.income_statement(period="quarter")
print(f"Income statement records: {len(income_statement)}")
# Display first 10 columns for brevity
print(income_statement.columns.tolist()[:10])

# Get cash flow statement (yearly)
cash_flow = finance.cash_flow(period="year")
print(f"Cash flow records: {len(cash_flow)}")

# Get financial ratios (MultiIndex DataFrame)
ratios = finance.ratio()
print(f"Ratio records: {len(ratios)}")
ratio_groups = ratios.columns.get_level_values(0).unique().tolist()
print(f"Ratio groups: {ratio_groups}")
# Groups: Meta, Capital Structure, Operational Efficiency,
#         Profitability, Liquidity, Valuation

# Export financial data to Excel
# This will create 'vci_financials.xlsx' in the current directory
with pd.ExcelWriter('vci_financials.xlsx') as writer:
    balance_sheet.to_excel(writer, sheet_name='Balance Sheet', index=False)
    income_statement.to_excel(writer, sheet_name='Income Statement', index=False)
    cash_flow.to_excel(writer, sheet_name='Cash Flow', index=False)

print("Financial data exported successfully")

```

--------------------------------

### Configure vnstock Retry Logic with Tenacity (Python)

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Configures the vnstock library's built-in retry mechanism using the 'tenacity' library. It sets the number of retries, backoff multiplier, minimum and maximum backoff times, and request timeout to enhance the robustness of API calls in production environments.

```python
from vnstock.config import Config

Config.RETRIES = 3
Config.BACKOFF_MULTIPLIER = 2
Config.BACKOFF_MIN = 1
Config.BACKOFF_MAX = 30
Config.TIMEOUT = 60
```

--------------------------------

### Analyze Shareholder Concentration and Company Events with vnstock

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

This snippet demonstrates how to analyze top shareholder ownership percentages, retrieve management officer details, track company events like dividends and stock splits, and access recent company news using the vnstock library. It relies on the 'shareholders' and 'company' objects from the library.

```python
from vnstock import Stock, Quote

# Assuming 'stock_code' is defined, e.g., stock_code = "VCI"
# Initialize Stock adapter (replace with actual initialization if needed)
# company = Stock(stock_code, source="vci") # This line might need adjustment based on actual library usage

# Example data structures (replace with actual vnstock objects)
class MockShareholders:
    def nlargest(self, n, column):
        class MockResult:
            def __init__(self):
                self.share_own_percent = [10.5, 8.2, 5.1, 4.3, 3.9, 3.5, 3.1, 2.8, 2.5, 2.2]
            def sum(self):
                return sum(self.share_own_percent)
        return MockResult()

class MockCompany:
    def shareholders(self):
        return MockShareholders()
    def officers(self, filter_by):
        return [
            {'officer_name': 'Nguyen Van A', 'officer_position': 'CEO', 'quantity': 10000},
            {'officer_name': 'Tran Thi B', 'officer_position': 'CFO', 'quantity': 5000},
            {'officer_name': 'Le Van C', 'officer_position': 'CTO', 'quantity': 0}
        ]
    def events(self):
        return [
            {'event_title': 'Dividend Payout', 'event_list_name': 'Cash Dividend', 'public_date': '2023-12-31'},
            {'event_title': 'Stock Split', 'event_list_name': '2:1 Split', 'public_date': '2024-01-15'}
        ]
    def news(self):
        return [
            {'news_title': 'Company Reports Strong Earnings', 'public_date': '2024-02-01'},
            {'news_title': 'New Product Launch Announced', 'public_date': '2024-02-05'}
        ]

# --- Actual Code Snippet --- 
# Initialize mock objects for demonstration if real objects are not available
# In a real scenario, you would initialize these from vnstock
shareholders = MockCompany().shareholders()
company = MockCompany() # Assuming company object holds methods for officers, events, news

# Analyze shareholder concentration
top_10_ownership = shareholders.nlargest(10, 'share_own_percent').sum()
print(f"Top 10 shareholders own: {top_10_ownership:.2f}%")

# Get management officers
officers = company.officers(filter_by="all")
# Convert to pandas DataFrame for easier manipulation if officers is a list of dicts
import pandas as pd
officers_df = pd.DataFrame(officers)
print(f"Total officers: {len(officers_df)}")
print(officers_df[['officer_name', 'officer_position', 'quantity']].head())

# Get insider ownership
insiders = officers_df[officers_df['quantity'] > 0].sort_values('quantity', ascending=False)
print(f"Officers with stock ownership: {len(insiders)}")

# Get company events (dividends, stock splits, IPOs)
events = company.events()
events_df = pd.DataFrame(events)
print(f"Total events: {len(events_df)}")
print(events_df[['event_title', 'event_list_name', 'public_date']].head())

# Get company news
news = company.news()
news_df = pd.DataFrame(news)
print(f"Recent news: {len(news_df)}")
print(news_df[['news_title', 'public_date']].head())

```

--------------------------------

### Fetch Stock Quote with Retry Logic (Python)

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Fetches historical stock quote data with an automatic retry mechanism. It includes data validation for OHLC consistency and error logging. Dependencies include the 'retry' library and 'vnstock.Quote'. Outputs a Pandas DataFrame or raises an error if fetching fails after retries.

```python
from vnstock import Quote
from tenacity import retry, stop_after_attempt, wait_exponential
import logging

logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def fetch_quote_with_retry(symbol, start, end):
    """Fetch quote data with automatic retry"""
    try:
        quote = Quote(source="vci", symbol=symbol)
        df = quote.history(start=start, end=end, resolution="1D")

        # Validate data
        if df is None or len(df) == 0:
            raise ValueError(f"No data returned for {symbol}")

        # Validate OHLC consistency
        invalid_rows = df[(df['high'] < df['low']) |
                         (df['high'] < df['open']) |
                         (df['high'] < df['close']) |
                         (df['low'] > df['open']) |
                         (df['low'] > df['close'])]

        if len(invalid_rows) > 0:
            logger.warning(f"Found {len(invalid_rows)} invalid OHLC rows")

        return df
    except Exception as e:
        logger.error(f"Error fetching {symbol}: {e}")
        raise

# Usage with error handling
symbols = ['VCI', 'ACB', 'BID', 'INVALID']

results = {}
for symbol in symbols:
    try:
        df = fetch_quote_with_retry(symbol, "2024-01-01", "2024-12-31")
        results[symbol] = df
        logger.info(f"Successfully fetched {symbol}: {len(df)} rows")
    except Exception as e:
        logger.error(f"Failed to fetch {symbol} after retries: {e}")
        results[symbol] = None
```

--------------------------------

### Cache Historical Stock Data with Pickle (Python)

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Implements a function to retrieve historical stock data, utilizing local caching with Python's pickle module to improve performance. If data for a given symbol, date range, and resolution is not found in the cache, it fetches it from the API and saves it for future use. It demonstrates caching for individual stock history and listing data.

```python
import pickle
import os
from vnstock import Quote, Listing
from vnstock.core.types import TimeFrame
from datetime import datetime

CACHE_DIR = 'cache'
os.makedirs(CACHE_DIR, exist_ok=True)

def get_cached_history(symbol, start, end, resolution="1D"):
    """Get historical data with caching"""
    cache_file = f'{CACHE_DIR}/{symbol}_{start}_{end}_{resolution}.pkl'

    if os.path.exists(cache_file):
        with open(cache_file, 'rb') as f:
            df = pickle.load(f)
        print(f"Loaded {symbol} from cache")
        return df

    quote = Quote(source="vci", symbol=symbol)
    df = quote.history(start=start, end=end, resolution=resolution)

    with open(cache_file, 'wb') as f:
        pickle.dump(df, f)
    print(f"Cached {symbol} for future use")

    return df

def get_cached_listing():
    """Get listing with caching"""
    cache_file = f'{CACHE_DIR}/listing_cache.pkl'

    if os.path.exists(cache_file):
        with open(cache_file, 'rb') as f:
            return pickle.load(f)

    listing = Listing(source="vci")
    data = listing.all_symbols(to_df=True)

    with open(cache_file, 'wb') as f:
        pickle.dump(data, f)

    return data

# Usage
df = get_cached_history("VCI", "2024-01-01", "2024-12-31", "1D")
print(df.head())

all_stocks = get_cached_listing()
print(f"Total stocks: {len(all_stocks)}")
```

--------------------------------

### Fetch Stock Quote with Fallback Sources (Python)

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Attempts to fetch stock quote data from a list of specified sources, falling back to the next source if one fails. It logs warnings for failed attempts and raises a ValueError if all sources fail. Dependencies include 'vnstock.Quote'. Outputs a tuple containing a Pandas DataFrame and the source name, or raises an error.

```python
from vnstock import Quote
import logging

logger = logging.getLogger(__name__)

def fetch_with_fallback(symbol, start, end, sources=["vci", "tcbs", "msn"]):
    """Try multiple sources with fallback"""
    for source in sources:
        try:
            quote = Quote(source=source, symbol=symbol)
            df = quote.history(start=start, end=end, resolution="1D")
            if df is not None and len(df) > 0:
                logger.info(f"Successfully fetched {symbol} from {source}")
                return df, source
        except Exception as e:
            logger.warning(f"Failed with {source}: {e}")
            continue

    raise ValueError(f"All sources failed for {symbol}")

# Test fallback
df, source = fetch_with_fallback("VCI", "2024-01-01", "2024-12-31")
print(f"Fetched from {source}: {len(df)} rows")
```

--------------------------------

### Implement Caching with Expiration for VNStock Data Fetching

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/03-custom-pipelines.md

This Python code implements a `CachedFetcher` class that extends `VNFetcher` to cache stock data locally. It automatically handles cache expiration based on a Time-To-Live (TTL) in hours, preventing redundant API calls and speeding up data retrieval. The cache is stored as pickle files, and the validity is checked by comparing the file's modification time with the TTL. If the cache is invalid or does not exist, it fetches fresh data and updates the cache.

```python
from vnstock_pipeline.template.vnstock import VNFetcher
from vnstock_data import Quote
import pandas as pd
import pickle
import os
from datetime import datetime, timedelta

class CachedFetcher(VNFetcher):
    """Fetch with local cache, automatic expiration"""
    
    def __init__(self, cache_dir: str = "./cache", ttl_hours: int = 24):
        self.cache_dir = cache_dir
        self.ttl_hours = ttl_hours
        os.makedirs(cache_dir, exist_ok=True)
    
    def _get_cache_path(self, ticker: str, start: str, end: str) -> str:
        filename = f"{ticker}_{start}_{end}.pkl"
        return os.path.join(self.cache_dir, filename)
    
    def _is_cache_valid(self, cache_path: str) -> bool:
        if not os.path.exists(cache_path):
            return False
        
        age = datetime.now() - datetime.fromtimestamp(os.path.getmtime(cache_path))
        return age < timedelta(hours=self.ttl_hours)
    
    def _vn_call(self, ticker: str, **kwargs) -> pd.DataFrame:
        start = kwargs.get("start", "2024-01-01")
        end = kwargs.get("end", "2024-12-02")
        cache_path = self._get_cache_path(ticker, start, end)
        
        # Check cache
        if self._is_cache_valid(cache_path):
            with open(cache_path, 'rb') as f:
                df = pickle.load(f)
                print(f"✅ {ticker} from cache")
                return df
        
        # Fetch fresh
        try:
            quote = Quote(source="vci", symbol=ticker)
            df = quote.history(start=start, end=end, interval=kwargs.get("interval", "1D"))
            
            # Save cache
            with open(cache_path, 'wb') as f:
                pickle.dump(df, f)
            
            print(f"✅ {ticker} freshly fetched")
            return df
        
        except Exception as e:
            print(f"❌ Error fetching {ticker}: {e}")
            return pd.DataFrame()

```

--------------------------------

### Production Data Export to CSV and Parquet (Python)

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/03-custom-pipelines.md

This class inherits from Exporter and handles saving data to both CSV and Parquet formats. It initializes by creating necessary directories for CSV and Parquet files. The export method saves the provided data for a given ticker, while the preview method allows reading and displaying the head of the CSV file for a ticker. It requires pandas and os modules.

```python
class ProductionExporter(Exporter):
    def __init__(self, base_path: str):
        self.base_path = base_path
        import os
        os.makedirs(f"{base_path}/csv", exist_ok=True)
        os.makedirs(f"{base_path}/parquet", exist_ok=True)
    
    def export(self, data, ticker: str, **kwargs):
        import os
        
        # CSV
        csv_path = os.path.join(self.base_path, "csv", f"{ticker}.csv")
        data.to_csv(csv_path, index=False)
        
        # Parquet
        parquet_path = os.path.join(self.base_path, "parquet", f"{ticker}.parquet")
        data.to_parquet(parquet_path, index=False, compression='snappy')
        
        logger.info(f"✅ Exported {ticker}")
    
    def preview(self, ticker: str, n: int = 5, **kwargs):
        import os
        csv_path = os.path.join(self.base_path, "csv", f"{ticker}.csv")
        if os.path.exists(csv_path):
            return pd.read_csv(csv_path).head(n)
        return None
```

--------------------------------

### Run vnstock Pipeline with Production Exporter in Python

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/02-tasks-and-builders.md

This code demonstrates how to instantiate and run the vnstock data processing pipeline. It defines a list of tickers, configures a Scheduler with specific components including the ProductionExporter, and then executes the pipeline for the given tickers and date range.

```python
if __name__ == "__main__":
    tickers = ['VCB', 'ACB', 'HPG', 'FPT', 'GAS']
    
    # Assuming Scheduler, RobustFetcher, StrictValidator, EnrichedTransformer are imported
    class Scheduler:
        def __init__(self, fetcher, validator, transformer, exporter, retry_attempts, backoff_factor):
            pass
        def run(self, tickers, fetcher_kwargs):
            pass
            
    class RobustFetcher: pass
    class StrictValidator: pass
    class EnrichedTransformer: pass

    scheduler = Scheduler(
        RobustFetcher(),
        StrictValidator(),
        EnrichedTransformer(),
        ProductionExporter("./production_data"),
        retry_attempts=3,
        backoff_factor=2.0
    )
    
    scheduler.run(
        tickers,
        fetcher_kwargs={
            "start": "2024-01-01",
            "end": "2024-12-02",
            "interval": "1D"
        }
    )
    
    logger("✅ Pipeline completed!")
```

--------------------------------

### Implement Caching for vnstock Fetcher in Python

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/02-tasks-and-builders.md

This demonstrates using a CachedFetcher to optimize data retrieval. By caching previously fetched data, it avoids redundant API calls, leading to faster execution and reduced load on the data source.

```python
# Assuming CachedFetcher is an imported class
# fetcher = CachedFetcher()
```

--------------------------------

### Export DataFrame to CSV and Excel (Python)

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Demonstrates how to export pandas DataFrames to different file formats. It shows saving a DataFrame to a CSV file and then exporting multiple DataFrames to different sheets within a single Excel file using pandas' ExcelWriter.

```python
import pandas as pd

# Export to CSV
df.to_csv('vci_daily.csv', index=False)
print("Exported to CSV")

# Export to Excel with multiple sheets
with pd.ExcelWriter('stock_analysis.xlsx') as writer:
    df.to_excel(writer, sheet_name='OHLCV', index=False)
    all_stocks.to_excel(writer, sheet_name='Listing', index=False)

print("Exported to Excel")
```

--------------------------------

### Filter Stocks by Finance Sector and HOSE Exchange (Python)

Source: https://context7.com/vinhson1987-sys/vnstock-agent-guide/llms.txt

Filters a pandas DataFrame to select stocks belonging to the 'Finance' industry and listed on the 'HOSE' exchange. It then prints the count and the first 10 entries of the filtered data, displaying the symbol and company name.

```python
finance_hose = merged[
    (merged['industry'] == 'Finance') &
    (merged['exchange'] == 'HOSE')
]
print(f"Finance stocks on HOSE: {len(finance_hose)}")
print(finance_hose[['symbol', 'company_name']].head(10))
```

--------------------------------

### Export Data to DuckDB for Querying

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/03-custom-pipelines.md

This Python class `DuckDBExporter` allows exporting pandas DataFrames to a DuckDB database. It handles table creation and insertion of data, enabling efficient querying of historical stock data by ticker symbol. The `query` method allows direct SQL queries against the exported data.

```python
from vnstock_pipeline.core.exporter import Exporter
import duckdb
import pandas as pd

class DuckDBExporter(Exporter):
    """Export to DuckDB for efficient querying"""
    
    def __init__(self, db_path: str = "stocks.duckdb"):
        self.db_path = db_path
        self.conn = duckdb.connect(db_path)
    
    def export(self, data, ticker: str, **kwargs):
        # Create table or append to existing
        table_name = f"stock_{ticker}".lower()
        
        try:
            self.conn.execute(f"""
                CREATE TABLE IF NOT EXISTS {table_name} AS
                SELECT * FROM data WHERE 1=0
            """)
        except:
            pass
        
        # Insert data
        self.conn.from_df(data).insert_into(table_name)
        print(f"✅ {ticker}: Inserted into {table_name}")
    
    def query(self, ticker: str, query: str):
        """Query data directly"""
        table_name = f"stock_{ticker}".lower()
        return self.conn.execute(
            f"SELECT * FROM {table_name} WHERE {query}"
        ).df()
    
    def __del__(self):
        if hasattr(self, 'conn'):
            self.conn.close()

# Usage
# exporter = DuckDBExporter("stocks.duckdb")

# Export
# for ticker in ['VCB', 'ACB']:
#     df = fetch_data(ticker) # Assuming fetch_data is defined elsewhere
#     exporter.export(df, ticker)

# Query
# recent_vcb = exporter.query("vcb", "time > '2024-12-01'")
# high_volume = exporter.query("vcb", "volume > 10000000")

```

--------------------------------

### Production Pipeline Components: Fetcher, Validator, Transformer with vnstock

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/02-tasks-and-builders.md

Demonstrates a production-ready data pipeline using vnstock components. It includes a RobustFetcher for data retrieval with logging, a StrictValidator to enforce data quality rules (minimum rows, valid high/low), and an EnrichedTransformer to add technical indicators like SMAs, volatility, and percentage change.

```python
from vnstock_pipeline.core.scheduler import Scheduler
from vnstock_pipeline.template.vnstock import VNFetcher, VNValidator, VNTransformer
from vnstock_pipeline.core.exporter import Exporter
import pandas as pd
import os
import logging

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 1. Custom Fetcher with retry
class RobustFetcher(VNFetcher):
    def _vn_call(self, ticker: str, **kwargs) -> pd.DataFrame:
        from vnstock_data import Quote
        quote = Quote(source="vnd", symbol=ticker)
        df = quote.history(
            start=kwargs.get("start", "2024-01-01"),
            end=kwargs.get("end", "2024-12-02"),
            interval=kwargs.get("interval", "1D")
        )
        logger.info(f"Fetched {len(df)} rows for {ticker}")
        return df

# 2. Strict Validator
class StrictValidator(VNValidator):
    required_columns = ["time", "open", "high", "low", "close", "volume"]
    
    def validate(self, data: pd.DataFrame) -> bool:
        if not super().validate(data):
            return False
        
        if len(data) < 10:  # Need at least 10 rows
            return False
        
        if (data['high'] < data['low']).any():
            return False
        
        return True

# 3. Enriched Transformer
class EnrichedTransformer(VNTransformer):
    def transform(self, data: pd.DataFrame) -> pd.DataFrame:
        df = super().transform(data)
        
        # Add indicators
        df['sma20'] = df['close'].rolling(20).mean()
        df['sma50'] = df['close'].rolling(50).mean()
        df['volatility_30d'] = df['close'].pct_change().rolling(30).std()
        df['change_pct'] = df['close'].pct_change() * 100
        
        logger.info(f"Enriched data with {len(df)} rows")
        return df
```

--------------------------------

### Multi-Source Data Fetching with vnstock

Source: https://github.com/vinhson1987-sys/vnstock-agent-guide/blob/main/docs/vnstock_pipeline/02-tasks-and-builders.md

Fetches stock data from multiple sources ('vci', 'vnd', 'cafef') and returns the first successful result. It uses the vnstock_data library and handles potential errors during fetching from each source. Returns an empty DataFrame if no source provides data.

```python
from vnstock_pipeline.template.vnstock import VNFetcher
import pandas as pd

class MultiSourceFetcher(VNFetcher):
    """Fetch từ nhiều nguồn, lấy source tốt nhất"""
    
    def _vn_call(self, ticker: str, **kwargs) -> pd.DataFrame:
        sources = ['vci', 'vnd', 'cafef']
        
        for source in sources:
            try:
                from vnstock_data import Quote
                quote = Quote(source=source, symbol=ticker)
                df = quote.history(
                    start=kwargs.get("start", "2024-01-01"),
                    end=kwargs.get("end", "2024-12-02"),
                    interval=kwargs.get("interval", "1D")
                )
                
                if len(df) > 0:
                    df['source'] = source
                    return df
            except:
                continue
        
        # Fallback: empty DataFrame
        return pd.DataFrame()
```