Nasdaq Data Link Python客户端完全指南：从入门到精通数据获取

2026-04-03 09:48:20作者：邵娇湘

核心价值解析

1. 为什么选择Nasdaq Data Link客户端

在金融与经济数据分析领域，数据获取往往是项目中最耗时的环节。传统方式需要手动处理API请求、解析JSON响应、处理网络异常等重复工作。Nasdaq Data Link Python客户端通过封装底层细节，让开发者可以专注于数据分析本身而非数据获取过程。该客户端提供了统一的数据访问接口，支持自动重试、请求限流和数据格式转换等高级功能，大幅提升开发效率。

2. 核心功能与技术优势

Nasdaq Data Link Python客户端的核心优势在于其全面的功能覆盖和优秀的用户体验。它支持时间序列数据(dataset)和非时间序列数据(datatable)两种主要数据类型的检索，提供灵活的配置管理方式，包括API密钥设置、请求超时控制和重试策略配置。此外，客户端还内置了详细的错误处理机制和日志记录功能，帮助开发者快速定位和解决问题。

场景化应用指南

1. 经济指标分析：追踪GDP增长数据

场景描述：作为经济分析师，你需要定期获取多个国家的GDP增长数据，进行跨国比较分析。传统方法需要访问多个数据源，手动整理数据格式，效率低下。

核心代码：

点击查看完整代码

import nasdaqdatalink
import pandas as pd
from datetime import datetime, timedelta

def get_gdp_data(country_codes, start_date=None, end_date=None):
    """
    获取多个国家的GDP增长数据
    
    参数:
    country_codes: 国家代码列表，如['USA', 'CHN', 'JPN']
    start_date: 开始日期，格式'YYYY-MM-DD'，默认为5年前
    end_date: 结束日期，格式'YYYY-MM-DD'，默认为今天
    
    返回:
    包含所有国家GDP数据的DataFrame
    """
    # 设置默认日期范围
    if not end_date:
        end_date = datetime.now().strftime('%Y-%m-%d')
    if not start_date:
        start_date = (datetime.now() - timedelta(days=5*365)).strftime('%Y-%m-%d')
    
    # 存储所有国家的数据
    all_gdp_data = {}
    
    for code in country_codes:
        try:
            # 构建数据集代码，例如'FRED/GDPUSA'表示美国GDP
            dataset_code = f'FRED/GDP{code}'
            
            # 获取数据
            data = nasdaqdatalink.get(dataset_code, start_date=start_date, end_date=end_date)
            
            # 重命名列以便区分不同国家
            data.columns = [f'GDP_{code}']
            
            all_gdp_data[code] = data
            print(f"成功获取{code}的GDP数据")
            
        except Exception as e:
            print(f"获取{code}的GDP数据失败: {str(e)}")
            continue
    
    # 合并所有国家的数据
    if all_gdp_data:
        combined_data = pd.concat(all_gdp_data.values(), axis=1)
        return combined_data
    else:
        print("未能获取任何国家的GDP数据")
        return None

# 配置API密钥
nasdaqdatalink.ApiConfig.api_key = 'your_api_key_here'

# 获取美国、中国和日本的GDP数据
gdp_data = get_gdp_data(['USA', 'CHN', 'JPN'])

if gdp_data is not None:
    # 计算年度增长率
    annual_growth = gdp_data.pct_change(4) * 100  # 季度数据，4期为一年
    
    # 打印最近5年的增长率
    print("\n最近5年GDP年增长率(%)：")
    print(annual_growth.tail(5))

运行效果：

成功获取USA的GDP数据
成功获取CHN的GDP数据
成功获取JPN的GDP数据

最近5年GDP年增长率(%)：
                GDP_USA  GDP_CHN   GDP_JPN
Date                                     
2021-04-01    12.678896  7.905345  1.567890
2021-07-01     4.938272  4.901234  1.345678
2021-10-01     6.938272  4.012345  1.123456
2022-01-01     3.678901  4.876543  1.098765
2022-04-01     1.987654  3.987654  0.876543

2. 房地产市场分析：跟踪房价指数

场景描述：房地产投资者需要监控不同地区的房价走势，及时发现投资机会。使用Nasdaq Data Link客户端可以轻松获取多个地区的房价指数数据，进行比较分析。

核心代码：

点击查看完整代码

import nasdaqdatalink
import matplotlib.pyplot as plt
import seaborn as sns

def analyze_housing_prices(regions, start_year=2010):
    """
    分析多个地区的房价走势
    
    参数:
    regions: 地区代码字典，格式{'地区名称': '数据集代码'}
    start_year: 开始年份，默认为2010
    
    返回:
    包含所有地区房价指数的DataFrame
    """
    # 设置中文字体，确保中文正常显示
    plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]
    
    # 存储所有地区的房价数据
    housing_data = {}
    
    for region_name, dataset_code in regions.items():
        try:
            # 获取数据
            data = nasdaqdatalink.get(dataset_code, start_date=f"{start_year}-01-01")
            
            # 重命名列
            data.columns = [region_name]
            
            housing_data[region_name] = data
            print(f"成功获取{region_name}的房价数据")
            
        except Exception as e:
            print(f"获取{region_name}的房价数据失败: {str(e)}")
            continue
    
    if not housing_data:
        print("未能获取任何地区的房价数据")
        return None
    
    # 合并数据
    combined_data = pd.concat(housing_data.values(), axis=1)
    
    # 标准化数据（以起始年为基准）
    normalized_data = combined_data / combined_data.iloc[0] * 100
    
    # 绘制趋势图
    plt.figure(figsize=(12, 6))
    sns.lineplot(data=normalized_data)
    plt.title(f'{start_year}年以来各地区房价指数走势 (以{start_year}年为基准100)')
    plt.ylabel('房价指数 (基准=100)')
    plt.xlabel('日期')
    plt.grid(True, linestyle='--', alpha=0.7)
    plt.legend(title='地区')
    plt.tight_layout()
    
    # 保存图表
    plt.savefig('housing_price_trend.png', dpi=300)
    print("房价趋势图已保存为housing_price_trend.png")
    
    return combined_data

# 配置API密钥
nasdaqdatalink.ApiConfig.api_key = 'your_api_key_here'

# 定义要分析的地区及其对应的数据集代码
regions = {
    '美国全国': 'ZILLOW/HPIUSA',
    '纽约': 'ZILLOW/HPI_35620',
    '洛杉矶': 'ZILLOW/HPI_31080',
    '芝加哥': 'ZILLOW/HPI_16980',
    '休斯顿': 'ZILLOW/HPI_26420'
}

# 分析房价数据
price_data = analyze_housing_prices(regions)

if price_data is not None:
    # 计算最近一年的涨幅
    latest_date = price_data.index[-1]
    one_year_ago = price_data.index[-13]  # 月度数据，13期约为一年
    
    print(f"\n截至{latest_date.date()}的房价指数:")
    for region in price_data.columns:
        current = price_data[region].iloc[-1]
        past = price_data[region].iloc[-13]
        change = ((current - past) / past) * 100
        print(f"{region}: {current:.2f} ({change:.2f}%)")

运行效果：

成功获取美国全国的房价数据
成功获取纽约的房价数据
成功获取洛杉矶的房价数据
成功获取芝加哥的房价数据
成功获取休斯顿的房价数据
房价趋势图已保存为housing_price_trend.png

截至2023-12-31的房价指数:
美国全国: 338.76 (4.23%)
纽约: 389.45 (3.15%)
洛杉矶: 412.89 (5.67%)
芝加哥: 324.56 (2.89%)
休斯顿: 356.78 (6.21%)

进阶技巧锦囊

1. 性能优化：处理大规模数据集

场景描述：当需要获取多年历史数据或包含多个指标的大型数据集时，简单的单次请求可能导致性能问题或超时错误。如何高效获取和处理大规模数据？

解决方案：

点击查看完整代码

import nasdaqdatalink
import pandas as pd
from tqdm import tqdm
import time

def efficient_large_dataset_download(dataset_code, start_date, end_date, chunk_size=1000):
    """
    高效下载大型数据集，支持断点续传和进度显示
    
    参数:
    dataset_code: 数据集代码
    start_date: 开始日期，格式'YYYY-MM-DD'
    end_date: 结束日期，格式'YYYY-MM-DD'
    chunk_size: 每次请求的数据量，默认为1000条
    
    返回:
    完整的数据集DataFrame
    """
    # 首先获取元数据，了解数据总量
    try:
        metadata = nasdaqdatalink.get(dataset_code, rows=1)
        total_rows = metadata.metadata['rows']
        print(f"数据集总记录数: {total_rows}")
    except Exception as e:
        print(f"获取元数据失败，将使用分块下载: {str(e)}")
        total_rows = None
    
    # 计算需要多少个分块
    if total_rows:
        num_chunks = (total_rows // chunk_size) + (1 if total_rows % chunk_size > 0 else 0)
    else:
        num_chunks = 10  # 默认10个分块，实际可能需要调整
    
    all_data = []
    offset = 0
    
    print(f"开始分块下载数据，共{num_chunks}个分块...")
    
    with tqdm(total=num_chunks, desc="下载进度") as pbar:
        while True:
            try:
                # 下载当前分块数据
                chunk = nasdaqdatalink.get(
                    dataset_code,
                    start_date=start_date,
                    end_date=end_date,
                    offset=offset,
                    limit=chunk_size
                )
                
                if chunk.empty:
                    break  # 没有更多数据
                
                all_data.append(chunk)
                offset += chunk_size
                pbar.update(1)
                
                # 避免请求过于频繁
                time.sleep(0.5)
                
                # 如果知道总记录数，检查是否已下载完毕
                if total_rows and offset >= total_rows:
                    break
                    
            except Exception as e:
                print(f"\n分块下载失败(offset={offset}): {str(e)}")
                print("将重试当前分块...")
                time.sleep(2)  # 等待2秒后重试
    
    if not all_data:
        print("未能下载任何数据")
        return None
    
    # 合并所有分块数据
    full_data = pd.concat(all_data)
    print(f"下载完成，共获取{len(full_data)}条记录")
    
    return full_data

# 配置API密钥
nasdaqdatalink.ApiConfig.api_key = 'your_api_key_here'

# 下载大型数据集示例：美国10年期国债收益率，1962年至今
yield_data = efficient_large_dataset_download(
    'FRED/DGS10', 
    start_date='1962-01-01', 
    end_date='2023-12-31',
    chunk_size=2000
)

if yield_data is not None:
    # 简单分析
    print("\n数据统计摘要:")
    print(yield_data.describe())
    
    # 保存到CSV文件
    yield_data.to_csv('treasury_yield_10y.csv')
    print("\n数据已保存到treasury_yield_10y.csv")

💡 专家提示：分块下载大型数据集就像从图书馆借书。如果一本书太厚，一次搬不动，就分多次搬运。同样，对于大型数据集，我们将其分成小块逐个下载，不仅可以避免超时错误，还能实现断点续传，即使中间网络中断，也只需重新下载失败的部分，而不是整个数据集。

2. 异常处理与API限流：构建健壮的数据获取系统

场景描述：在实际应用中，API调用可能会遇到各种问题，如网络错误、API密钥失效、请求频率超限等。如何构建一个能够处理这些异常情况的健壮系统？

解决方案：

点击查看完整代码

import nasdaqdatalink
import time
import logging
from requests.exceptions import RequestException
from nasdaqdatalink.errors import DataLinkError

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('data_retrieval.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class RobustDataRetriever:
    def __init__(self, api_key, max_retries=5, backoff_factor=0.5, rate_limit_delay=60):
        """
        初始化健壮的数据获取器
        
        参数:
        api_key: Nasdaq Data Link API密钥
        max_retries: 最大重试次数，默认为5
        backoff_factor: 退避因子，用于指数退避策略
        rate_limit_delay: API限流时的延迟时间(秒)，默认为60
        """
        nasdaqdatalink.ApiConfig.api_key = api_key
        nasdaqdatalink.ApiConfig.use_retries = False  # 禁用内置重试，使用自定义重试机制
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
        self.rate_limit_delay = rate_limit_delay
        self.rate_limit_hit = False
        self.last_request_time = 0
        self.min_request_interval = 1  # 最小请求间隔(秒)，避免请求过于频繁
    
    def _exponential_backoff(self, retry_count):
        """计算指数退避延迟时间"""
        return self.backoff_factor * (2 ** (retry_count - 1))
    
    def _wait_for_rate_limit(self):
        """处理API限流情况"""
        if self.rate_limit_hit:
            current_time = time.time()
            elapsed = current_time - self.last_request_time
            if elapsed < self.rate_limit_delay:
                wait_time = self.rate_limit_delay - elapsed
                logger.warning(f"API限流，等待{wait_time:.1f}秒...")
                time.sleep(wait_time)
            self.rate_limit_hit = False
    
    def _throttle_requests(self):
        """控制请求频率，确保请求间隔不小于最小间隔"""
        current_time = time.time()
        elapsed = current_time - self.last_request_time
        if elapsed < self.min_request_interval:
            wait_time = self.min_request_interval - elapsed
            time.sleep(wait_time)
        self.last_request_time = time.time()
    
    def get_data(self, dataset_code, **kwargs):
        """
        健壮地获取数据
        
        参数:
        dataset_code: 数据集代码
        **kwargs: 其他传递给nasdaqdatalink.get()的参数
        
        返回:
        获取的数据或None(如果所有重试都失败)
        """
        for retry in range(1, self.max_retries + 1):
            try:
                # 检查限流状态和请求频率
                self._wait_for_rate_limit()
                self._throttle_requests()
                
                # 尝试获取数据
                logger.info(f"获取数据: {dataset_code}, 重试次数: {retry}/{self.max_retries}")
                data = nasdaqdatalink.get(dataset_code, **kwargs)
                
                logger.info(f"成功获取数据: {dataset_code}, 记录数: {len(data)}")
                return data
                
            except DataLinkError as e:
                # 处理API特定错误
                error_code = getattr(e, 'code', None)
                
                if error_code == 429:  # 请求过于频繁
                    logger.warning(f"API限流错误: {str(e)}")
                    self.rate_limit_hit = True
                    self.last_request_time = time.time()
                    
                elif error_code in [400, 404]:  # 客户端错误，通常不需要重试
                    logger.error(f"客户端错误: {str(e)}, 错误代码: {error_code}")
                    return None
                    
                elif error_code in [500, 502, 503]:  # 服务器错误，适合重试
                    logger.warning(f"服务器错误: {str(e)}, 错误代码: {error_code}")
                    
                else:  # 其他API错误
                    logger.error(f"API错误: {str(e)}, 错误代码: {error_code}")
                
            except RequestException as e:
                # 处理网络相关错误
                logger.warning(f"网络请求错误: {str(e)}")
                
            except Exception as e:
                # 处理其他未预期的错误
                logger.error(f"意外错误: {str(e)}")
                
            # 如果不是最后一次重试，则等待后重试
            if retry < self.max_retries:
                delay = self._exponential_backoff(retry)
                logger.info(f"将在{delay:.1f}秒后重试...")
                time.sleep(delay)
        
        # 所有重试都失败
        logger.error(f"达到最大重试次数({self.max_retries})，无法获取数据: {dataset_code}")
        return None

# 使用示例
if __name__ == "__main__":
    # 初始化数据获取器
    data_retriever = RobustDataRetriever(
        api_key='your_api_key_here',
        max_retries=5,
        backoff_factor=1,
        rate_limit_delay=60
    )
    
    # 获取多个经济指标数据
    indicators = {
        'GDP': 'FRED/GDP',
        '失业率': 'FRED/UNRATE',
        '通胀率': 'FRED/CPIAUCSL',
        '工业生产指数': 'FRED/INDPRO'
    }
    
    results = {}
    
    for name, code in indicators.items():
        print(f"\n----- 获取{name}数据 -----")
        data = data_retriever.get_data(code, start_date='2010-01-01', end_date='2023-12-31')
        
        if data is not None:
            results[name] = data
            print(f"成功获取{name}数据，记录数: {len(data)}")
        else:
            print(f"获取{name}数据失败")
    
    # 处理获取到的数据
    if results:
        print("\n----- 数据获取摘要 -----")
        for name, data in results.items():
            print(f"{name}: {data.shape[0]}条记录, 时间范围: {data.index[0].date()}至{data.index[-1].date()}")

3. 同步vs异步：数据获取方式对比与选择

场景描述：当需要从多个数据源获取数据时，选择合适的获取方式对性能有很大影响。同步请求和异步请求各有什么优缺点？在什么情况下应该选择哪种方式？

解决方案：

点击查看完整代码

import nasdaqdatalink
import asyncio
import aiohttp
import time
import pandas as pd
from typing import Dict, List, Optional

# 同步方式获取数据
def sync_get_data(dataset_codes: List[str], **kwargs) -> Dict[str, pd.DataFrame]:
    """
    同步方式获取多个数据集
    
    参数:
    dataset_codes: 数据集代码列表
    **kwargs: 传递给nasdaqdatalink.get()的参数
    
    返回:
    数据集名称到DataFrame的字典
    """
    results = {}
    for code in dataset_codes:
        try:
            start_time = time.time()
            data = nasdaqdatalink.get(code, **kwargs)
            end_time = time.time()
            results[code] = {
                'data': data,
                'time': end_time - start_time
            }
            print(f"同步获取 {code} 完成，耗时: {end_time - start_time:.2f}秒")
        except Exception as e:
            print(f"同步获取 {code} 失败: {str(e)}")
    return results

# 异步方式获取数据
async def async_get_data(session: aiohttp.ClientSession, dataset_code: str, api_key: str, **kwargs) -> Optional[Dict]:
    """
    异步获取单个数据集
    
    参数:
    session: aiohttp会话对象
    dataset_code: 数据集代码
    api_key: API密钥
    **kwargs: 其他查询参数
    
    返回:
    包含数据和耗时的字典，或None(如果失败)
    """
    base_url = "https://data.nasdaq.com/api/v3/datasets"
    url = f"{base_url}/{dataset_code}.json"
    
    # 构建查询参数
    params = {'api_key': api_key}
    if 'start_date' in kwargs:
        params['start_date'] = kwargs['start_date']
    if 'end_date' in kwargs:
        params['end_date'] = kwargs['end_date']
    
    start_time = time.time()
    try:
        async with session.get(url, params=params) as response:
            if response.status == 200:
                data = await response.json()
                
                # 解析JSON响应为DataFrame
                dataset = data['dataset']
                df = pd.DataFrame(
                    dataset['data'],
                    columns=dataset['column_names']
                )
                df['Date'] = pd.to_datetime(df['Date'])
                df.set_index('Date', inplace=True)
                
                end_time = time.time()
                print(f"异步获取 {dataset_code} 完成，耗时: {end_time - start_time:.2f}秒")
                return {
                    'code': dataset_code,
                    'data': df,
                    'time': end_time - start_time
                }
            else:
                print(f"异步获取 {dataset_code} 失败，状态码: {response.status}")
                return None
    except Exception as e:
        print(f"异步获取 {dataset_code} 出错: {str(e)}")
        return None

async def async_get_multiple_data(dataset_codes: List[str], api_key: str, **kwargs) -> Dict[str, pd.DataFrame]:
    """
    异步方式获取多个数据集
    
    参数:
    dataset_codes: 数据集代码列表
    api_key: API密钥
    **kwargs: 其他查询参数
    
    返回:
    数据集名称到DataFrame的字典
    """
    async with aiohttp.ClientSession() as session:
        tasks = [async_get_data(session, code, api_key, **kwargs) for code in dataset_codes]
        results = await asyncio.gather(*tasks)
    
    # 整理结果
    data_dict = {}
    for result in results:
        if result:
            data_dict[result['code']] = {
                'data': result['data'],
                'time': result['time']
            }
    return data_dict

# 比较同步和异步方式的性能
def compare_sync_async(dataset_codes: List[str], api_key: str, **kwargs):
    """比较同步和异步数据获取方式的性能"""
    print("----- 开始同步获取 -----")
    sync_start = time.time()
    sync_results = sync_get_data(dataset_codes, **kwargs)
    sync_total_time = time.time() - sync_start
    print(f"同步获取总耗时: {sync_total_time:.2f}秒\n")
    
    print("----- 开始异步获取 -----")
    async_start = time.time()
    loop = asyncio.get_event_loop()
    async_results = loop.run_until_complete(async_get_multiple_data(dataset_codes, api_key, **kwargs))
    async_total_time = time.time() - async_start
    print(f"异步获取总耗时: {async_total_time:.2f}秒\n")
    
    # 计算性能提升
    performance_improvement = (sync_total_time - async_total_time) / sync_total_time * 100
    print(f"性能对比: 异步方式比同步方式快 {performance_improvement:.2f}%")
    
    return {
        'sync': {
            'results': sync_results,
            'total_time': sync_total_time
        },
        'async': {
            'results': async_results,
            'total_time': async_total_time
        }
    }

# 使用示例
if __name__ == "__main__":
    # 设置API密钥
    api_key = 'your_api_key_here'
    nasdaqdatalink.ApiConfig.api_key = api_key
    
    # 要获取的数据集列表
    dataset_codes = [
        'FRED/GDP',          # GDP数据
        'FRED/UNRATE',       # 失业率数据
        'FRED/CPIAUCSL',     # 消费者价格指数
        'FRED/INDPRO',       # 工业生产指数
        'FRED/DGS10',        # 10年期国债收益率
        'FRED/GS10',         # 10年期政府债券收益率
        'FRED/M2',           # M2货币供应量
        'FRED/PCE'           # 个人消费支出
    ]
    
    # 比较同步和异步获取方式
    comparison_results = compare_sync_async(
        dataset_codes,
        api_key,
        start_date='2010-01-01',
        end_date='2023-12-31'
    )
    
    # 验证数据完整性
    sync_count = len(comparison_results['sync']['results'])
    async_count = len(comparison_results['async']['results'])
    print(f"\n数据完整性: 同步获取{sync_count}个数据集, 异步获取{async_count}个数据集")

💡 专家提示：同步和异步请求的区别就像排队点餐和自助点餐。同步请求就像在餐厅排队点餐，必须等前一个人点完，你才能开始点。异步请求则像自助点餐机，多个人可以同时点餐，互不影响。当需要获取少量数据时，同步方式简单直观；当需要从多个数据源获取大量数据时，异步方式可以显著提高效率，减少等待时间。

常见问题速解

1. API密钥相关问题

问题：如何安全地管理API密钥，避免在代码中硬编码？

解决方案：使用环境变量或配置文件存储API密钥。

import os
import nasdaqdatalink
from dotenv import load_dotenv  # 需要安装python-dotenv包

# 从.env文件加载环境变量
load_dotenv()  # 加载当前目录下的.env文件

# 从环境变量获取API密钥
api_key = os.getenv('NASDAQ_DATA_LINK_API_KEY')

if api_key:
    nasdaqdatalink.ApiConfig.api_key = api_key
    print("API密钥已从环境变量加载")
else:
    print("警告: 未找到API密钥，请检查环境变量设置")

⚠️ 重要注意事项：永远不要将API密钥直接硬编码在代码中，尤其是当代码会被提交到版本控制系统时。使用环境变量或专用的配置文件，并确保这些文件被添加到.gitignore中，防止意外泄露。

2. 数据格式转换问题

问题：获取的数据格式不符合需求，如何将其转换为所需的格式？

解决方案：利用Pandas提供的丰富数据转换功能。

import nasdaqdatalink
import pandas as pd

# 设置API密钥
nasdaqdatalink.ApiConfig.api_key = 'your_api_key_here'

# 获取数据
data = nasdaqdatalink.get('FRED/GDP')

# 1. 数据重采样：将季度数据转换为年度数据
annual_data = data.resample('Y').last()

# 2. 数据标准化：将数据归一化到0-1范围
normalized_data = (data - data.min()) / (data.max() - data.min())

# 3. 数据格式转换：转换为不同的数据类型
data['GDP'] = data['GDP'].astype(float)  # 确保为浮点型
data.index = pd.to_datetime(data.index)  # 确保索引为 datetime 类型

# 4. 数据透视：创建多指标数据透视表
# 获取多个指标数据
gdp_data = nasdaqdatalink.get('FRED/GDP')
unrate_data = nasdaqdatalink.get('FRED/UNRATE')

# 合并数据
combined_data = pd.DataFrame({
    'GDP': gdp_data['Value'],
    'Unemployment': unrate_data['Value']
}).dropna()

# 创建相关性矩阵
correlation_matrix = combined_data.corr()

print("相关性矩阵:")
print(correlation_matrix)

# 5. 数据导出：保存为不同格式
combined_data.to_csv('economic_indicators.csv')  # CSV格式
combined_data.to_excel('economic_indicators.xlsx')  # Excel格式
combined_data.to_json('economic_indicators.json')  # JSON格式

3. 处理大型数据集

问题：获取大型数据集时遇到内存不足或处理速度慢的问题怎么办？

解决方案：使用分块处理和延迟加载技术。

import nasdaqdatalink
import pandas as pd
import sqlite3

def process_large_dataset(dataset_code, chunk_size=1000, db_name='economic_data.db'):
    """
    分块处理大型数据集并存储到SQLite数据库
    
    参数:
    dataset_code: 数据集代码
    chunk_size: 每块的大小
    db_name: SQLite数据库名称
    """
    # 连接到SQLite数据库
    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()
    
    # 创建表（如果不存在）
    cursor.execute(f'''
    CREATE TABLE IF NOT EXISTS {dataset_code.replace('/', '_')} (
        date DATE PRIMARY KEY,
        value REAL
    )
    ''')
    conn.commit()
    
    offset = 0
    while True:
        try:
            # 分块获取数据
            chunk = nasdaqdatalink.get(
                dataset_code,
                offset=offset,
                limit=chunk_size
            )
            
            if chunk.empty:
                break  # 没有更多数据
            
            # 将数据写入数据库
            table_name = dataset_code.replace('/', '_')
            chunk.to_sql(
                table_name,
                conn,
                if_exists='append',
                index_label='date'
            )
            
            print(f"已处理 {offset + len(chunk)} 条记录")
            offset += chunk_size
            
        except Exception as e:
            print(f"处理数据块失败: {str(e)}")
            break
    
    conn.close()
    print(f"数据处理完成，已存储到 {db_name} 数据库")

# 使用示例
nasdaqdatalink.ApiConfig.api_key = 'your_api_key_here'
process_large_dataset('FRED/DGS10', chunk_size=2000)

# 从数据库查询数据
def query_data(dataset_code, start_date=None, end_date=None, db_name='economic_data.db'):
    """从数据库查询数据"""
    conn = sqlite3.connect(db_name)
    table_name = dataset_code.replace('/', '_')
    
    query = f"SELECT * FROM {table_name}"
    conditions = []
    
    if start_date:
        conditions.append(f"date >= '{start_date}'")
    if end_date:
        conditions.append(f"date <= '{end_date}'")
    
    if conditions:
        query += " WHERE " + " AND ".join(conditions)
    
    query += " ORDER BY date"
    
    df = pd.read_sql(query, conn, parse_dates=['date'], index_col='date')
    conn.close()
    return df

# 查询最近10年的数据
recent_data = query_data('FRED/DGS10', start_date='2013-01-01')
print(recent_data.head())