mirrors2/stock-tracker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

stock-tracker

Browse Source
This commit is contained in:
C菌
2026-07-04 00:17:11 +08:00
commit 6087341a48
6463 changed files with 1929869 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.venv/lib/python3.9/site-packages/akshare/other/__init__.py
+6
View File
@@ -0,0 +1,6 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
Date: 2021/5/14 17:52
Desc:
"""
backend/.venv/lib/python3.9/site-packages/akshare/other/other_car_cpca.py
+934
View File
@@ -0,0 +1,934 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
Date: 2024/8/3 20:00
Desc: 乘联会
http://data.cpcadata.com/FuelMarket
"""
import pandas as pd
import requests
def car_market_total_cpca(
symbol: str = "狭义乘用车", indicator: str = "产量"
) -> pd.DataFrame:
"""
乘联会-统计数据-总体市场
http://data.cpcadata.com/TotalMarket
:param symbol: choice of {"狭义乘用车", "广义乘用车"}
:type symbol: str
:param indicator: choice of {"产量", "批发", "零售", "出口"}
:type indicator: str
:return: 统计数据-总体市场
:rtype: pandas.DataFrame
"""
url = "http://data.cpcadata.com/api/chartlist"
params = {"charttype": "1"}
r = requests.get(url, params=params)
data_json = r.json()
big_df = pd.DataFrame()
if symbol == "狭义乘用车":
temp_df = pd.DataFrame(data_json[0]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[0]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
if indicator == "产量":
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 0], temp_previous_year_df.iloc[:, 0]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif indicator == "批发":
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 1], temp_previous_year_df.iloc[:, 1]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif indicator == "零售":
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 2], temp_previous_year_df.iloc[:, 2]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif indicator == "出口":
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 3], temp_previous_year_df.iloc[:, 3]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
else:
temp_df = pd.DataFrame(data_json[1]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[1]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
if indicator == "产量":
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 0], temp_previous_year_df.iloc[:, 0]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif indicator == "批发":
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 1], temp_previous_year_df.iloc[:, 1]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif indicator == "零售":
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 2], temp_previous_year_df.iloc[:, 2]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif indicator == "出口":
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 3], temp_previous_year_df.iloc[:, 3]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
return big_df
def __car_market_man_rank_cpca_pifa(symbol: str = "狭义乘用车-累计") -> pd.DataFrame:
"""
乘联会-统计数据-厂商排名
http://data.cpcadata.com/ManRank
:param symbol: choice of {"狭义乘用车-单月", "狭义乘用车-累计", "广义乘用车-单月", "广义乘用车-累计"}
:type symbol: str
:return: 统计数据-厂商排名
:rtype: pandas.DataFrame
"""
url = "http://data.cpcadata.com/api/chartlist"
params = {"charttype": "2"}
r = requests.get(url, params=params)
data_json = r.json()
big_df = pd.DataFrame()
if symbol == "狭义乘用车-累计":
temp_df = pd.DataFrame(data_json[0]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[0]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 0], temp_previous_year_df.iloc[:, 0]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["厂商"] = temp_df["厂商"]
big_df = big_df[
[
"厂商",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "狭义乘用车-单月":
temp_df = pd.DataFrame(data_json[1]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[1]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 0], temp_previous_year_df.iloc[:, 0]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["厂商"] = temp_df["厂商"]
big_df = big_df[
[
"厂商",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "广义乘用车-累计":
temp_df = pd.DataFrame(data_json[2]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[2]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 0], temp_previous_year_df.iloc[:, 0]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["厂商"] = temp_df["厂商"]
big_df = big_df[
[
"厂商",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "广义乘用车-单月":
temp_df = pd.DataFrame(data_json[3]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[3]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 0], temp_previous_year_df.iloc[:, 0]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["厂商"] = temp_df["厂商"]
big_df = big_df[
[
"厂商",
temp_df.columns[2],
temp_df.columns[1],
]
]
return big_df
def __car_market_man_rank_cpca_lingshou(
symbol: str = "狭义乘用车-累计",
) -> pd.DataFrame:
"""
乘联会-统计数据-厂商排名
http://data.cpcadata.com/ManRank
:param symbol: choice of {"狭义乘用车-单月", "狭义乘用车-累计", "广义乘用车-单月", "广义乘用车-累计"}
:type symbol: str
:return: 统计数据-厂商排名
:rtype: pandas.DataFrame
"""
url = "http://data.cpcadata.com/api/chartlist_2"
params = {"charttype": "2"}
r = requests.get(url, params=params)
data_json = r.json()
big_df = pd.DataFrame()
if symbol == "狭义乘用车-累计":
temp_df = pd.DataFrame(data_json[0]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[0]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 1], temp_previous_year_df.iloc[:, 1]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["厂商"] = temp_df["厂商"]
big_df = big_df[
[
"厂商",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "狭义乘用车-单月":
temp_df = pd.DataFrame(data_json[1]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[1]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 1], temp_previous_year_df.iloc[:, 1]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["厂商"] = temp_df["厂商"]
big_df = big_df[
[
"厂商",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "广义乘用车-累计":
temp_df = pd.DataFrame(data_json[2]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[2]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 1], temp_previous_year_df.iloc[:, 1]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["厂商"] = temp_df["厂商"]
big_df = big_df[
[
"厂商",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "广义乘用车-单月":
temp_df = pd.DataFrame(data_json[3]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[3]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 1], temp_previous_year_df.iloc[:, 1]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["厂商"] = temp_df["厂商"]
big_df = big_df[
[
"厂商",
temp_df.columns[2],
temp_df.columns[1],
]
]
return big_df
def car_market_man_rank_cpca(
symbol: str = "狭义乘用车-单月", indicator: str = "批发"
) -> pd.DataFrame:
"""
乘联会-统计数据-厂商排名
http://data.cpcadata.com/ManRank
:param symbol: choice of {"狭义乘用车-单月", "狭义乘用车-累计", "广义乘用车-单月", "广义乘用车-累计"}
:type symbol: str
:param indicator: choice of {"批发", "零售"}
:type indicator: str
:return: 统计数据-厂商排名
:rtype: pandas.DataFrame
"""
if indicator == "批发":
temp_df = __car_market_man_rank_cpca_pifa(symbol=symbol)
return temp_df
else:
temp_df = __car_market_man_rank_cpca_lingshou(symbol=symbol)
return temp_df
def __car_market_cate_cpca_pifa(symbol: str = "MPV") -> pd.DataFrame:
"""
乘联会-统计数据-车型大类
http://data.cpcadata.com/CategoryMarket
:param symbol: choice of {"轿车", "MPV", "SUV", "占比"}
:type symbol: str
:return: 统计数据-车型大类
:rtype: pandas.DataFrame
"""
url = "http://data.cpcadata.com/api/chartlist"
params = {"charttype": "3"}
r = requests.get(url, params=params)
data_json = r.json()
big_df = pd.DataFrame()
if symbol == "MPV":
temp_df = pd.DataFrame(data_json[0]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[0]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[1]])
try:
temp_current_year_list.append(item[temp_df.columns[2]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 1], temp_previous_year_df.iloc[:, 1]]
).T
big_df.columns = [temp_df.columns[2], temp_df.columns[1]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "SUV":
temp_df = pd.DataFrame(data_json[1]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[1]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[1]])
try:
temp_current_year_list.append(item[temp_df.columns[2]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 1], temp_previous_year_df.iloc[:, 1]]
).T
big_df.columns = [temp_df.columns[2], temp_df.columns[1]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "轿车":
temp_df = pd.DataFrame(data_json[2]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[2]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[1]])
try:
temp_current_year_list.append(item[temp_df.columns[2]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 1], temp_previous_year_df.iloc[:, 1]]
).T
big_df.columns = [temp_df.columns[2], temp_df.columns[1]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[1],
temp_df.columns[2],
]
]
elif symbol == "占比":
temp_df = pd.DataFrame(data_json[3]["dataList"])
temp_mpv_year_list = []
temp_suv_year_list = []
temp_jiaoche_year_list = []
for item in data_json[3]["dataList"]:
temp_mpv_year_list.append(item[temp_df.columns[1]])
try:
temp_suv_year_list.append(item[temp_df.columns[2]])
temp_jiaoche_year_list.append(item[temp_df.columns[3]])
except: # noqa: E722
continue
temp_mpv_year_df = pd.DataFrame(temp_mpv_year_list)
temp_suv_year_df = pd.DataFrame(temp_suv_year_list)
temp_jiaoche_year_df = pd.DataFrame(temp_jiaoche_year_list)
big_df = pd.DataFrame(
[
temp_mpv_year_df.iloc[:, 2],
temp_suv_year_df.iloc[:, 2],
temp_jiaoche_year_df.iloc[:, 2],
]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2], temp_df.columns[3]]
big_df["月份"] = temp_df["月份"]
big_df = big_df[
["月份", temp_df.columns[1], temp_df.columns[2], temp_df.columns[3]]
]
return big_df
def __car_market_cate_cpca_lingshou(
symbol: str = "狭义乘用车-累计",
) -> pd.DataFrame:
"""
乘联会-统计数据-车型大类
http://data.cpcadata.com/CategoryMarket
:param symbol: choice of {"轿车", "MPV", "SUV", "占比"}
:type symbol: str
:return: 统计数据-车型大类
:rtype: pandas.DataFrame
"""
url = "http://data.cpcadata.com/api/chartlist"
params = {"charttype": "3"}
r = requests.get(url, params=params)
data_json = r.json()
big_df = pd.DataFrame()
if symbol == "MPV":
temp_df = pd.DataFrame(data_json[0]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[0]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[1]])
try:
temp_current_year_list.append(item[temp_df.columns[2]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 2], temp_previous_year_df.iloc[:, 2]]
).T
big_df.columns = [temp_df.columns[2], temp_df.columns[1]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "SUV":
temp_df = pd.DataFrame(data_json[1]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[1]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[1]])
try:
temp_current_year_list.append(item[temp_df.columns[2]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 2], temp_previous_year_df.iloc[:, 2]]
).T
big_df.columns = [temp_df.columns[2], temp_df.columns[1]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "轿车":
temp_df = pd.DataFrame(data_json[2]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[2]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[1]])
try:
temp_current_year_list.append(item[temp_df.columns[2]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 2], temp_previous_year_df.iloc[:, 2]]
).T
big_df.columns = [temp_df.columns[2], temp_df.columns[1]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "占比":
temp_df = pd.DataFrame(data_json[3]["dataList"])
temp_mpv_year_list = []
temp_suv_year_list = []
temp_jiaoche_year_list = []
for item in data_json[3]["dataList"]:
temp_mpv_year_list.append(item[temp_df.columns[1]])
try:
temp_suv_year_list.append(item[temp_df.columns[2]])
temp_jiaoche_year_list.append(item[temp_df.columns[3]])
except: # noqa: E722
continue
temp_mpv_year_df = pd.DataFrame(temp_mpv_year_list)
temp_suv_year_df = pd.DataFrame(temp_suv_year_list)
temp_jiaoche_year_df = pd.DataFrame(temp_jiaoche_year_list)
big_df = pd.DataFrame(
[
temp_mpv_year_df.iloc[:, 3],
temp_suv_year_df.iloc[:, 3],
temp_jiaoche_year_df.iloc[:, 3],
]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2], temp_df.columns[3]]
big_df["月份"] = temp_df["月份"]
big_df = big_df[
["月份", temp_df.columns[1], temp_df.columns[2], temp_df.columns[3]]
]
return big_df
def car_market_cate_cpca(symbol: str = "轿车", indicator: str = "批发") -> pd.DataFrame:
"""
乘联会-统计数据-车型大类
http://data.cpcadata.com/CategoryMarket
:param symbol: choice of {"轿车", "MPV", "SUV", "占比"}
:type symbol: str
:param indicator: choice of {"批发", "零售"}
:type indicator: str
:return: 统计数据-车型大类
:rtype: pandas.DataFrame
"""
if indicator == "批发":
temp_df = __car_market_cate_cpca_pifa(symbol=symbol)
return temp_df
else:
temp_df = __car_market_cate_cpca_lingshou(symbol=symbol)
return temp_df
def car_market_country_cpca() -> pd.DataFrame:
"""
乘联会-统计数据-国别细分市场
http://data.cpcadata.com/CountryMarket
:return: 统计数据-车型大类
:rtype: pandas.DataFrame
"""
url = "http://data.cpcadata.com/api/chartlist"
params = {"charttype": "4"}
r = requests.get(url=url, params=params)
data_json = r.json()
temp_df = pd.DataFrame(data_json[0]["dataList"])
for item in temp_df.columns[1:]:
temp_list = []
for item_list in temp_df[item]:
temp_list.append(item_list[2])
temp_df[item] = pd.to_numeric(temp_list, errors="coerce")
return temp_df
def car_market_segment_cpca(symbol: str = "轿车") -> pd.DataFrame:
"""
乘联会-统计数据-级别细分市场
http://data.cpcadata.com/SegmentMarket
:param symbol: choice of {"轿车", "MPV", "SUV"}
:type symbol: str
:return: 统计数据-车型大类
:rtype: pandas.DataFrame
"""
url = "http://data.cpcadata.com/api/chartlist"
params = {"charttype": "5"}
r = requests.get(url=url, params=params)
data_json = r.json()
if symbol == "MPV":
temp_df = pd.DataFrame(data_json[0]["dataList"])
for item in temp_df.columns[1:]:
temp_list = []
for item_list in temp_df[item]:
temp_list.append(item_list[2])
temp_df[item] = pd.to_numeric(temp_list, errors="coerce")
elif symbol == "SUV":
temp_df = pd.DataFrame(data_json[1]["dataList"])
for item in temp_df.columns[1:]:
temp_list = []
for item_list in temp_df[item]:
temp_list.append(item_list[2])
temp_df[item] = pd.to_numeric(temp_list, errors="coerce")
else:
temp_df = pd.DataFrame(data_json[2]["dataList"])
for item in temp_df.columns[1:]:
temp_list = []
for item_list in temp_df[item]:
temp_list.append(item_list[2])
temp_df[item] = pd.to_numeric(temp_list, errors="coerce")
return temp_df
def car_market_fuel_cpca(symbol: str = "整体市场") -> pd.DataFrame:
"""
乘联会-统计数据-新能源细分市场
:param symbol: choice of {"整体市场", "销量占比-PHEV-BEV", "销量占比-ICE-NEV"}
:type symbol: str
https://data.cpcadata.com/FuelMarket
:return: 新能源细分市场
:rtype: pandas.DataFrame
"""
url = "http://data.cpcadata.com/api/chartlist"
params = {"charttype": "6"}
r = requests.get(url, params=params)
data_json = r.json()
if symbol == "整体市场":
temp_df = pd.DataFrame(data_json[0]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[0]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 2], temp_previous_year_df.iloc[:, 2]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["month"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
elif symbol == "销量占比-PHEV-BEV":
temp_df = pd.DataFrame(data_json[1]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[1]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 2], temp_previous_year_df.iloc[:, 2]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["月份"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
else:
temp_df = pd.DataFrame(data_json[2]["dataList"])
temp_current_year_list = []
temp_previous_year_list = []
for item in data_json[2]["dataList"]:
temp_previous_year_list.append(item[temp_df.columns[2]])
try:
temp_current_year_list.append(item[temp_df.columns[1]])
except: # noqa: E722
continue
temp_current_year_df = pd.DataFrame(temp_current_year_list)
temp_previous_year_df = pd.DataFrame(temp_previous_year_list)
big_df = pd.DataFrame(
[temp_current_year_df.iloc[:, 2], temp_previous_year_df.iloc[:, 2]]
).T
big_df.columns = [temp_df.columns[1], temp_df.columns[2]]
big_df["月份"] = temp_df["月份"]
big_df = big_df[
[
"月份",
temp_df.columns[2],
temp_df.columns[1],
]
]
return big_df
if __name__ == "__main__":
car_market_total_cpca_df = car_market_total_cpca(
symbol="狭义乘用车", indicator="产量"
)
print(car_market_total_cpca_df)
car_market_total_cpca_df = car_market_total_cpca(
symbol="广义乘用车", indicator="产量"
)
print(car_market_total_cpca_df)
car_market_total_cpca_df = car_market_total_cpca(
symbol="狭义乘用车", indicator="批发"
)
print(car_market_total_cpca_df)
car_market_total_cpca_df = car_market_total_cpca(
symbol="广义乘用车", indicator="批发"
)
print(car_market_total_cpca_df)
car_market_total_cpca_df = car_market_total_cpca(
symbol="狭义乘用车", indicator="零售"
)
print(car_market_total_cpca_df)
car_market_total_cpca_df = car_market_total_cpca(
symbol="广义乘用车", indicator="零售"
)
print(car_market_total_cpca_df)
car_market_total_cpca_df = car_market_total_cpca(
symbol="狭义乘用车", indicator="出口"
)
print(car_market_total_cpca_df)
car_market_total_cpca_df = car_market_total_cpca(
symbol="广义乘用车", indicator="出口"
)
print(car_market_total_cpca_df)
car_market_man_rank_cpca_df = car_market_man_rank_cpca(
symbol="狭义乘用车-单月", indicator="批发"
)
print(car_market_man_rank_cpca_df)
car_market_man_rank_cpca_df = car_market_man_rank_cpca(
symbol="狭义乘用车-累计", indicator="批发"
)
print(car_market_man_rank_cpca_df)
car_market_man_rank_cpca_df = car_market_man_rank_cpca(
symbol="广义乘用车-单月", indicator="批发"
)
print(car_market_man_rank_cpca_df)
car_market_man_rank_cpca_df = car_market_man_rank_cpca(
symbol="广义乘用车-累计", indicator="批发"
)
print(car_market_man_rank_cpca_df)
car_market_man_rank_cpca_df = car_market_man_rank_cpca(
symbol="狭义乘用车-单月", indicator="零售"
)
print(car_market_man_rank_cpca_df)
car_market_man_rank_cpca_df = car_market_man_rank_cpca(
symbol="狭义乘用车-累计", indicator="零售"
)
print(car_market_man_rank_cpca_df)
car_market_man_rank_cpca_df = car_market_man_rank_cpca(
symbol="广义乘用车-单月", indicator="零售"
)
print(car_market_man_rank_cpca_df)
car_market_man_rank_cpca_df = car_market_man_rank_cpca(
symbol="广义乘用车-累计", indicator="零售"
)
print(car_market_man_rank_cpca_df)
car_market_cate_cpca_df = car_market_cate_cpca(symbol="轿车", indicator="批发")
print(car_market_cate_cpca_df)
car_market_cate_cpca_df = car_market_cate_cpca(symbol="MPV", indicator="批发")
print(car_market_cate_cpca_df)
car_market_cate_cpca_df = car_market_cate_cpca(symbol="SUV", indicator="批发")
print(car_market_cate_cpca_df)
car_market_cate_cpca_df = car_market_cate_cpca(symbol="占比", indicator="批发")
print(car_market_cate_cpca_df)
car_market_cate_cpca_df = car_market_cate_cpca(symbol="轿车", indicator="零售")
print(car_market_cate_cpca_df)
car_market_cate_cpca_df = car_market_cate_cpca(symbol="MPV", indicator="零售")
print(car_market_cate_cpca_df)
car_market_cate_cpca_df = car_market_cate_cpca(symbol="SUV", indicator="零售")
print(car_market_cate_cpca_df)
car_market_cate_cpca_df = car_market_cate_cpca(symbol="占比", indicator="零售")
print(car_market_cate_cpca_df)
car_market_country_cpca_df = car_market_country_cpca()
print(car_market_country_cpca_df)
car_market_segment_cpca_df = car_market_segment_cpca(symbol="轿车")
print(car_market_segment_cpca_df)
car_market_segment_cpca_df = car_market_segment_cpca(symbol="MPV")
print(car_market_segment_cpca_df)
car_market_segment_cpca_df = car_market_segment_cpca(symbol="SUV")
print(car_market_segment_cpca_df)
car_market_fuel_cpca_df = car_market_fuel_cpca(symbol="整体市场")
print(car_market_fuel_cpca_df)
car_market_fuel_cpca_df = car_market_fuel_cpca(symbol="销量占比-PHEV-BEV")
print(car_market_fuel_cpca_df)
car_market_fuel_cpca_df = car_market_fuel_cpca(symbol="销量占比-ICE-NEV")
print(car_market_fuel_cpca_df)
backend/.venv/lib/python3.9/site-packages/akshare/other/other_car_gasgoo.py
+84
View File
@@ -0,0 +1,84 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
Date: 2024/4/3 19:00
Desc: 汽车行业制造企业数据库
http://i.gasgoo.com/data/ranking
"""
import pandas as pd
import requests
from akshare.utils import demjson
def car_sale_rank_gasgoo(symbol: str = "车企榜", date: str = "202109") -> pd.DataFrame:
"""
盖世汽车-汽车行业制造企业数据库-销量数据
https://i.gasgoo.com/data/ranking
:param symbol: choice of {"车企榜", "品牌榜", "车型榜"}
:type symbol: str
:param date: 查询的年份和月份
:type date: str
:return: 销量数据
:rtype: pandas.DataFrame
"""
symbol_map = {
"车型榜": "M",
"车企榜": "F",
"品牌榜": "B",
}
url = "https://i.gasgoo.com/data/sales/AutoModelSalesRank.aspx/GetSalesRank"
payload = {
"countryID": "",
"endM": str(int(date[4:6])),
"endY": date[:4],
"energy": "",
"modelGradeID": "",
"modelTypeID": "",
"orderBy": f"{date[:4]}-{str(int(date[4:6]))}",
"queryDate": f"{date[:4]}-{str(int(date[4:6]))}",
"rankType": symbol_map[symbol],
"startY": date[:4],
"startM": str(int(date[4:6])),
}
headers = {
"Accept": "application/json, text/javascript, */*; q=0.01",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"Content-Length": "195",
"Content-Type": "application/json; charset=UTF-8",
"Cookie": "Hm_lvt_8e90480b1bf68ede548c407057660718=1636981448; _ga=GA1.2.858318653.1636981449; "
"_gid=GA1.2.1448165285.1636981449; _gat=1; Hm_lpvt_8e90480b1bf68ede548c407057660718=1636982578",
"Host": "i.gasgoo.com",
"Origin": "https://i.gasgoo.com",
"Pragma": "no-cache",
"Referer": "https://i.gasgoo.com/data/sales/AutoModelSalesRank.aspx/GetSalesRank",
"sec-ch-ua": '"Google Chrome";v="95", "Chromium";v="95", ";Not A Brand";v="99"',
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Windows"',
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/95.0.4638.69 Safari/537.36",
"X-Requested-With": "XMLHttpRequest",
}
r = requests.post(url, json=payload, headers=headers)
data_json = r.json()
data_json = demjson.decode(data_json["d"])
temp_df = pd.DataFrame(data_json)
return temp_df
if __name__ == "__main__":
car_sale_rank_gasgoo_df = car_sale_rank_gasgoo(symbol="品牌榜", date="202311")
print(car_sale_rank_gasgoo_df)
car_sale_rank_gasgoo_df = car_sale_rank_gasgoo(symbol="车型榜", date="202311")
print(car_sale_rank_gasgoo_df)
car_sale_rank_gasgoo_df = car_sale_rank_gasgoo(symbol="车企榜", date="202311")
print(car_sale_rank_gasgoo_df)
backend/.venv/lib/python3.9/site-packages/akshare/other/other_taptap.py
+170
View File
@@ -0,0 +1,170 @@
# -*- coding: utf-8 -*-
"""
Date: 2026/05/02
Desc: TapTap 游戏榜单数据接口
"""
import re
import time
from typing import Optional
import pandas as pd
import requests
# ============================================================
# 常量配置
# ============================================================
_TAPTAP_BASE_URL = "https://www.taptap.cn/webapiv2/app-top/v2/hits"
_TAPTAP_HEADERS = {
"User-Agent": (
"Mozilla/5.0 (iPhone; CPU iPhone OS 18_5 like Mac OS X) "
"AppleWebKit/605.1.15 (KHTML, like Gecko) "
"Version/18.0 Mobile/15E148 Safari/604.1"
),
"Referer": "https://www.taptap.cn/",
"Accept": "application/json, text/plain, */*",
}
_TAPTAP_X_UA = (
"V=1&PN=WebM&LANG=zh_CN&VN_CODE=102&LOC=CN&PLT=iOS&DS=Android"
"&UID=12f0a48b-bd25-4dce-9d50-27924e83da1d&OS=iOS&OSV=18.5"
)
# 内部参数（不对外暴露）
_TAPTAP_TIMEOUT = 15.0
_TAPTAP_SLEEP = 0.4
_TAPTAP_PAGE_SIZE = 10
_TAPTAP_MAX_LOOPS = 200
# 榜单类型枚举
_TAPTAP_RANK_TYPE_MAP = {
"热玩榜": "pop",
"热门榜": "hot",
"新品榜": "new",
"预约榜": "reserve",
"热卖榜": "sell",
}
def _clean_html(text: Optional[str]) -> str:
"""清洗 HTML 标签和实体字符"""
if text is None or (isinstance(text, float) and pd.isna(text)):
return ""
text = str(text)
text = re.sub(r"<br[^>]*/?>", "\n", text)
text = re.sub(r"<[^>]+>", "", text)
replacements = {
"&amp;": "&", "&lt;": "<", "&gt;": ">",
"&#34;": '"', "&#39;": "'", "&quot;": '"', "&nbsp;": " ",
}
for k, v in replacements.items():
text = text.replace(k, v)
return re.sub(r"\n{3,}", "\n\n", text).strip()
def game_hot_rank_taptap(symbol: str = "热玩榜") -> pd.DataFrame:
"""
TapTap-游戏榜单
https://www.taptap.cn/top/played
:param symbol: 榜单类型; 可选 {"热玩榜", "热门榜", "新品榜", "预约榜", "热卖榜"}
:type symbol: str
:return: 游戏榜单数据
:rtype: pandas.DataFrame
"""
if symbol not in _TAPTAP_RANK_TYPE_MAP:
raise ValueError(
f"symbol 仅支持 {list(_TAPTAP_RANK_TYPE_MAP.keys())}, 当前传入: {symbol}"
)
type_name = _TAPTAP_RANK_TYPE_MAP[symbol]
session = requests.Session()
session.headers.update(_TAPTAP_HEADERS)
all_games = []
total: Optional[int] = None
offset = 0
for _ in range(_TAPTAP_MAX_LOOPS):
params = {
"from": offset,
"limit": _TAPTAP_PAGE_SIZE,
"type_name": type_name,
"X-UA": _TAPTAP_X_UA,
}
r = session.get(_TAPTAP_BASE_URL, params=params, timeout=_TAPTAP_TIMEOUT)
r.raise_for_status()
js = r.json()
if not js.get("success"):
raise RuntimeError(f"TapTap 接口返回失败: {js}")
data = js.get("data", {}) or {}
page_list = data.get("list", []) or []
if total is None:
total = data.get("total", 0)
if not page_list:
break
all_games.extend(page_list)
if total and len(all_games) >= total:
break
offset += _TAPTAP_PAGE_SIZE
time.sleep(_TAPTAP_SLEEP)
if not all_games:
return pd.DataFrame()
df = pd.json_normalize(all_games)
df["标签"] = [
", ".join([t.get("value", "") for t in (item.get("app", {}).get("tags") or [])])
for item in all_games
]
rename_map = {
"app.id": "游戏ID",
"app.title": "游戏名称",
"app.icon.url": "图标链接",
"app.stat.rating.score": "评分",
"app.stat.hits_total": "总点击量",
"app.stat.play_total": "游玩次数",
"app.stat.review_count": "评论数",
"app.stat.fans_count": "粉丝数",
"app.description.text": "简介",
"app.rec_text": "推荐语",
"app.released_time": "发布时间戳",
}
df.rename(columns=rename_map, inplace=True)
if "评分" in df.columns:
df["评分"] = pd.to_numeric(df["评分"], errors="coerce").astype("float64")
for col in ["总点击量", "游玩次数", "评论数", "粉丝数"]:
if col in df.columns:
df[col] = pd.to_numeric(df[col], errors="coerce").astype("Int64")
if "发布时间戳" in df.columns:
df["发布时间"] = pd.to_datetime(
df["发布时间戳"], unit="s", errors="coerce"
)
for col in ["游戏名称", "游戏ID", "图标链接", "推荐语", "标签"]:
if col in df.columns:
df[col] = df[col].astype("string").str.strip()
if "简介" in df.columns:
df["简介"] = df["简介"].apply(_clean_html).astype("string")
columns_order = [
"游戏名称", "评分", "总点击量", "游玩次数", "评论数",
"粉丝数", "标签", "推荐语", "发布时间",
"游戏ID", "图标链接", "简介",
]
df = df[[c for c in columns_order if c in df.columns]]
df = df.drop_duplicates(subset=["游戏ID"]).reset_index(drop=True)
df.insert(0, "排名", df.index + 1)
return df
if __name__ == "__main__":
game_hot_rank_taptap_df = game_hot_rank_taptap(symbol="热玩榜")
print(game_hot_rank_taptap_df)
game_hot_rank_taptap_df = game_hot_rank_taptap(symbol="热门榜")
print(game_hot_rank_taptap_df)
game_hot_rank_taptap_df = game_hot_rank_taptap(symbol="新品榜")
print(game_hot_rank_taptap_df)
game_hot_rank_taptap_df = game_hot_rank_taptap(symbol="预约榜")
print(game_hot_rank_taptap_df)
game_hot_rank_taptap_df = game_hot_rank_taptap(symbol="热卖榜")
print(game_hot_rank_taptap_df)