Search API

from valyu import Valyu

valyu = Valyu()

response = valyu.search(
    "What are the latest developments in quantum computing?"
)

print(f"Found {len(response.results)} results")
for result in response.results:
    print(f"Title: {result.title}")
    print(f"URL: {result.url}")
    print(f"Content: {result.content[:200]}...")

class SearchResponse:
    success: bool
    error: Optional[str]
    tx_id: str
    query: str
    results: List[SearchResult]
    results_by_source: ResultsBySource
    total_deduction_dollars: float
    total_characters: int

class SearchResult:
    title: str
    url: str
    content: str
    description: Optional[str]
    source: str
    price: float
    length: int
    relevance_score: float
    data_type: Optional[str]  # "structured" | "unstructured"
    # Additional fields for academic/proprietary sources
    publication_date: Optional[str]
    authors: Optional[List[str]]
    citation: Optional[str]
    citation_count: Optional[int]
    doi: Optional[str]
    references: Optional[str]
    metadata: Optional[Dict[str, Any]]

response = valyu.search(query, fast_mode=True)

# Web search only
web_response = valyu.search(query, search_type="web", max_num_results=10)

# Proprietary datasets only
proprietary_response = valyu.search(query, search_type="proprietary", max_num_results=8)

# Both web and proprietary (default)
all_response = valyu.search(query, search_type="all", max_num_results=12)

response = valyu.search(
    "quantum computing applications",
    search_type="all",
    max_num_results=10,
    included_sources=["valyu/valyu-arxiv", "valyu/valyu-pubmed", "valyu/valyu-biorxiv"],
    response_length="medium"
)

response = valyu.search(
    "quantum computing applications",
    search_type="all",
    max_num_results=10,
    excluded_sources=["example.com", "example.org"],
    response_length="medium"
)

response = valyu.search(
    "climate change research",
    search_type="web",
    source_biases={
        "nasa.gov": 5,
        "noaa.gov": 3,
        "epa.gov": 2,
        "nih.gov": 1,
        "example.com": -4
    }
)

response = valyu.search(
    "climate data",
    included_sources=["nasa.gov", "noaa.gov", "epa.gov"],
    source_biases={
        "nasa.gov": 5,
        "epa.gov": -3
    }
)

source_biases={
    "nih.gov": 2,               # General NIH content boosted
    "nih.gov/research": -3      # Specific section demoted
}

response = valyu.search(
    "renewable energy policies",
    country_code="US",
    start_date="2024-01-01",
    end_date="2024-12-31",
    max_num_results=7,
    category="government policy"
)

# Predefined lengths
short_response = valyu.search(query, response_length="short")  # 25k characters
medium_response = valyu.search(query, response_length="medium")  # 50k characters
large_response = valyu.search(query, response_length="large")  # 100k characters

# Custom character limit
custom_response = valyu.search(query, response_length=15000)  # Custom limit

def academic_research(query: str):
    response = valyu.search(
        query,
        search_type="proprietary",
        included_sources=["valyu/valyu-pubmed", "valyu/valyu-arxiv"],
        max_num_results=15,
        response_length="large",
        category="academic research"
    )

    if response.success:
        print(f"=== Academic Research Results ===")
        print(f"Found {len(response.results)} papers for: \"{query}\"")
        
        # Group by source
        arxiv_papers = [r for r in response.results if "arxiv" in r.source]
        pubmed_papers = [r for r in response.results if "pubmed" in r.source]
        
        print(f"\nArXiv Papers: {len(arxiv_papers)}")
        for i, paper in enumerate(arxiv_papers, 1):
            print(f"{i}. {paper.title}")
            print(f"   Relevance: {paper.relevance_score:.2f}")
            print(f"   URL: {paper.url}")
            if paper.publication_date:
                print(f"   Published: {paper.publication_date}")
            if paper.authors:
                print(f"   Authors: {', '.join(paper.authors)}")
        
        print(f"\nPubMed Articles: {len(pubmed_papers)}")
        for i, article in enumerate(pubmed_papers, 1):
            print(f"{i}. {article.title}")
            print(f"   Relevance: {article.relevance_score:.2f}")
            if article.citation:
                print(f"   Citation: {article.citation}")
        
        return {
            "arxiv": arxiv_papers,
            "pubmed": pubmed_papers,
            "query": query
        }
    
    return None

# Usage examples - dates are included in natural language
covid_research = academic_research(
    "COVID-19 vaccine efficacy studies published between 2023 and 2024"
)

ai_research = academic_research(
    "recent machine learning breakthroughs in the last 2 years"
)

climate_research = academic_research(
    "climate change mitigation strategies peer-reviewed research since 2020"
)

def financial_intelligence(query: str, analysis_type: str):
    """analysis_type: 'fundamental', 'technical', or 'news'"""
    sources = {
        "fundamental": ["valyu/valyu-stocks", "sec.gov", "sec.gov"],
        "technical": ["valyu/valyu-stocks", "kaggle.com", "yahoo.com"],
        "news": ["treasury.gov", "federalreserve.gov", "sec.gov", "fred.stlouisfed.org"]
    }

    response = valyu.search(
        query,
        search_type="all" if analysis_type in ["fundamental", "technical"] else "web",
        included_sources=sources[analysis_type],
        max_num_results=10,
        response_length="medium",
        category="financial analysis"
    )

    if response.success:
        print(f"=== {analysis_type.upper()} Analysis ===")
        print(f"Query: \"{query}\"")
        
        for i, result in enumerate(response.results, 1):
            print(f"\n{i}. {result.title}")
            print(f"   Source: {result.source}")
            print(f"   Relevance: {result.relevance_score:.2f}")
            print(f"   URL: {result.url}")
            
            # Show excerpt for financial data
            if len(result.content) > 200:
                print(f"   Preview: {result.content[:200]}...")
            
            if result.publication_date:
                print(f"   Date: {result.publication_date}")
        
        return {
            "results": response.results,
            "analysis_type": analysis_type,
            "query": query
        }
    
    return None

# Usage examples - include timeframes in natural language
tesla_fundamentals = financial_intelligence(
    "Tesla financial performance Q3 2024 earnings revenue profit margins", 
    "fundamental"
)

apple_news = financial_intelligence(
    "Apple latest news this week product announcements stock updates", 
    "news"
)

bitcoin_technical = financial_intelligence(
    "Bitcoin price analysis technical indicators support resistance levels recent trends",
    "technical"
)

from datetime import datetime, timedelta

def news_monitoring(queries: List[str], days_back: int = 7, country: str = "US"):
    """
    Monitor multiple news topics with date and country filtering.
    Requires API key with increased_max_results permission for >20 results.
    """
    end_date = datetime.now().strftime("%Y-%m-%d")
    start_date = (datetime.now() - timedelta(days=days_back)).strftime("%Y-%m-%d")
    
    all_results = []
    
    for query in queries:
        response = valyu.search(
            query,
            search_type="news",  # Use news mode
            max_num_results=50,  # Up to 100 with increased_max_results permission
            start_date=start_date,
            end_date=end_date,
            country_code=country,
            response_length="short"
        )

        if response.success:
            all_results.append({
                "query": query,
                "articles": response.results,
                "count": len(response.results)
            })

    # Generate monitoring report
    print(f"=== News Monitoring Report ===")
    print(f"Date range: {start_date} to {end_date}")
    print(f"Country: {country}")
    print(f"Monitoring {len(queries)} topics\n")
    
    for result in all_results:
        query, articles, count = result["query"], result["articles"], result["count"]
        print(f"📰 {query.upper()}: {count} articles")
        
        for i, article in enumerate(articles[:5], 1):
            print(f"   {i}. {article.title}")
            print(f"      Date: {article.publication_date or 'N/A'}")
            print(f"      URL: {article.url}")
        print("")

    return all_results

# Monitor tech news from the last 7 days in the US
tech_news = news_monitoring(
    queries=[
        "artificial intelligence breakthroughs",
        "quantum computing progress", 
        "cryptocurrency regulation"
    ],
    days_back=7,
    country="US"
)

# Monitor business news from the last 30 days
business_news = news_monitoring(
    queries=[
        "Tesla earnings report",
        "Federal Reserve interest rate decisions",
        "tech layoffs announcements"
    ],
    days_back=30,
    country="US"
)

response = valyu.search("test query", max_num_results=5)

if not response.success:
    print("Search failed:", response.error)
    
    # Handle specific error cases
    if "insufficient credits" in response.error:
        print("Please add more credits to your account")
    elif "invalid" in response.error:
        print("Check your search parameters")
    
    return

# Process successful results
print(f"Transaction ID: {response.tx_id}")

for index, result in enumerate(response.results):
    print(f"{index + 1}. {result.title}")
    print(f"   Relevance: {result.relevance_score}")
    print(f"   Source: {result.source}")
    print(f"   URL: {result.url}")

import asyncio
from valyu import AsyncValyu

async def main():
    async with AsyncValyu() as valyu:
        response = await valyu.search(
            "What are the latest developments in quantum computing?"
        )
        for result in response.results:
            print(result.title, "—", result.url)

asyncio.run(main())

response = await valyu.search(
    "Real options valuation binomial Monte Carlo natural resource R&D",
    included_sources=["wiley/wiley-finance-papers", "wiley/wiley-finance-books"],
    max_num_results=15,
)

import asyncio
from valyu import AsyncValyu

queries = [
    "DCF valuation terminal value",
    "Black-Scholes options pricing",
    "GARCH volatility modelling",
    "Monte Carlo exotic derivatives",
]

async def main():
    async with AsyncValyu() as valyu:
        responses = await asyncio.gather(*[
            valyu.search(q, max_num_results=10) for q in queries
        ])
        for q, r in zip(queries, responses):
            print(f"{q}: {len(r.results)} results")

asyncio.run(main())