02 · Data Acquisition in Practice

import akshare as ak

 # Kweichow Moutai daily bars (forward adjusted); symbols/fields per latest akshare docs
df = ak.stock_zh_a_hist(
    symbol="600519",
    period="daily",
    start_date="20200101",
    end_date="20241231",
    adjust="qfq",          # forward adjustment, see Section 4
)

df.to_csv("data/600519_daily_qfq.csv", index=False, encoding="utf-8-sig")
print(df.head())
print(df.tail())

import akshare as ak

# Income statement / key financial indicators (interface names may change between versions; check docs)
fin = ak.stock_financial_abstract(symbol="600519")
print(fin.head())

import tushare as ts

ts.set_token("your_token")            # get after registering at tushare.pro
pro = ts.pro_api()

# Daily bars (unadjusted); ts_code format: code.market
df = pro.daily(ts_code="600519.SH", start_date="20200101", end_date="20241231")
df.to_csv("data/600519_daily_tushare.csv", index=False)
print(df.head())

import akshare as ak

# RB0 = rebar dominant continuous (symbol rule: product code + 0 means dominant), period swappable
df = ak.futures_main_sina(symbol="RB0", start_date="20230101", end_date="20241231")
print(df.columns)
df.to_csv("data/rb_main_daily.csv", index=False, encoding="utf-8-sig")

# For a single contract (e.g., rebar RB2501 delivering January 2025)
df_single = ak.futures_zh_daily_sina(symbol="RB2501")
print(df_single.head())

import time
import requests
import pandas as pd

BASE_URL = "https://api.binance.com/api/v3"   # spot; futures use fapi.binance.com, per official docs

def fetch_klines(symbol, interval, start_ms, end_ms, limit=1000):
    """Fetch a range of klines; mind rate limits: Binance bills by weight, see below"""
    params = {
        "symbol": symbol,
        "interval": interval,          # 1m/1h/1d/1w...
        "startTime": start_ms,
        "endTime": end_ms,
        "limit": limit,                # max 1000 candles per request
    }
    r = requests.get(f"{BASE_URL}/klines", params=params, timeout=15)
    r.raise_for_status()
    # Watch remaining weight; slow down automatically when near the cap
    used = int(r.headers.get("x-mbx-used-weight-1m", 0))
    if used > 2000:                    # default limit 6000/min, keep headroom
        time.sleep(30)
    time.sleep(0.1)                    # base throttle
    return r.json()

def klines_to_df(raw):
    cols = ["open_time", "open", "high", "low", "close", "volume",
            "close_time", "quote_vol", "trades", "taker_buy_base",
            "taker_buy_quote", "ignore"]
    df = pd.DataFrame(raw, columns=cols)
    for c in ["open", "high", "low", "close", "volume"]:
        df[c] = df[c].astype(float)
    df["open_time"] = pd.to_datetime(df["open_time"], unit="ms", utc=True)
    return df[["open_time", "open", "high", "low", "close", "volume"]]

# Page through 1h klines by time, 1000 at a time
start = int(pd.Timestamp("2024-01-01", tz="UTC").timestamp() * 1000)
end = start + 1000 * 3600 * 1000       # 1000 one-hour candles
raw = fetch_klines("BTCUSDT", "1h", start, end)
df = klines_to_df(raw)
df.to_csv("data/btcusdt_1h.csv", index=False)
print(df.head())

import pandas as pd

df = pd.read_csv("data/600519_daily_qfq.csv", parse_dates=["日期"])

# Unify column names for later processing (AKShare Chinese columns)
df = df.rename(columns={"日期": "date", "开盘": "open", "收盘": "close",
                        "最高": "high", "最低": "low", "成交量": "volume"})
df["volume"] = df["volume"].astype(float)

# Near-zero volume → suspended or anomalous day: keep prices, flag it
df.loc[df["volume"] < 1, "halted"] = True
df.loc[df["volume"] >= 1, "halted"] = False
print(df[df["halted"] == True].head())   # sanity-check the number of halted days

# Fetch backward-adjusted data for backtesting (fields/params per latest akshare docs)
df_hfq = ak.stock_zh_a_hist(symbol="600519", period="daily",
                            start_date="20200101", end_date="20241231",
                            adjust="hfq")
df_hfq.to_csv("data/600519_daily_hfq.csv", index=False)

import datetime as dt
import pandas as pd
import akshare as ak

def incremental_update(symbol="600519", path="data/600519_daily_qfq.csv"):
    try:
        old = pd.read_csv(path, parse_dates=["日期"])
        last = old["日期"].max()
        start = (last + dt.timedelta(days=1)).strftime("%Y%m%d")
    except FileNotFoundError:
        old, start = None, "20200101"

    new = ak.stock_zh_a_hist(symbol=symbol, period="daily",
                             start_date=start, end_date=dt.date.today().strftime("%Y%m%d"),
                             adjust="qfq")
    if new.empty:
        return
    df = pd.concat([old, new], ignore_index=True).drop_duplicates(subset="日期")
    df.to_csv(path, index=False, encoding="utf-8-sig")

incremental_update()

import sqlite3

con = sqlite3.connect("data/market.db")
# Create table: open_time as primary key, skip duplicates on write (INSERT OR IGNORE)
con.execute("""CREATE TABLE IF NOT EXISTS kline_1h (
    open_time TEXT PRIMARY KEY, open REAL, high REAL,
    low REAL, close REAL, volume REAL)""")
df.to_sql("kline_1h", con, if_exists="append", index=False)  # teaching simplification: use INSERT OR IGNORE in production
con.commit()

# Quick health-check script
assert (df["high"] >= df["low"]).all(), "high < low anomaly"
assert (df["close"].dropna() > 0).all(), "non-positive price"
assert df["date"].is_unique, "duplicate dates found"
assert df["date"].is_monotonic_increasing, "dates not ascending"
print("Basic checks passed")