查询股票创新低且底部有资金进场的python

一、技术栈（按层次划分）

1. 数据获取层

2. 数据处理层

3. 终端展示层

4. 输出与持久化层

5. 工程化实践

二、功能拆解（从输入到输出）

🎯 核心目标

在"今日创历史新低"的股票池中，找出"当日主力资金净流入"的标的——即技术面创新低但资金面有承接的"底部抄底"信号。

📋 完整功能流程

text

┌─────────────────────────────────────────────────────────┐
│ ① 启动 & 标题展示 │
│ Panel 显示标题 + 当前运行时间 │
└─────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ ② 拉取创新低股票池 │
│ ak.stock_rank_cxd_ths("历史新低") │
│ → 得到 [股票代码, 股票简称] 列表 │
│ → status 转圈提示；空数据友好退出 │
└─────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ ③ 逐只核查资金流（带进度条） │
│ 对每只股票: │
│ a. 按代码前缀判断市场（6=sh, 0/3=sz, 8/4=bj） │
│ b. ak.stock_individual_fund_flow 拉取资金流 │
│ c. 取最新一行，提取: │
│ – 主力净流入-净额 │
│ – 超大单净流入-净额 │
│ – 主力净流入-净占比 │
│ – 收盘价 / 涨跌幅 / 日期 │
│ d. 筛选: 主力净流入 > 0 │
│ e. time.sleep(0.3) 限速 │
│ f. 异常计入 errors，不中断 │
└─────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ ④ 结果美化渲染 │
│ – 按主力净流入降序排序 │
│ – Rich 表格：序号/代码/名称/收盘价/涨跌幅/ │
│ 主力净流入/超大单净流入/主力净占比/日期 │
│ – 涨跌幅红涨绿跌；金额自动转"万/亿" │
│ – 空结果给出友好提示 Panel │
└─────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ ⑤ 汇总统计面板 │
│ – 符合条件标的数 │
│ – 主力净流入合计 │
│ – 超大单净流入合计 │
│ – 主力净占比均值 │
│ – 数据获取失败数 │
└─────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ ⑥ 导出 Excel │
│ 带时间戳的 .xlsx，失败给黄色警告不中断 │
└─────────────────────────────────────────────────────────┘

🧩 特色功能点

🚀 可扩展方向（脚本预留的接口）

• 阈值参数化：main_net > 1000_0000 筛选"千万级大资金"

• 多日确认：检查近 3 日主力连续净流入，过滤单日脉冲

• 换手率融合：低换手 + 主力流入 = 更隐蔽的吸筹

• 推送集成：结果转 HTML 表格推送到钉钉/邮件，做成盘后定时任务

• 回测验证：将该信号作为因子，统计历史次日/次周胜率

三、一句话总结

这是一个"技术面创新低 × 资金面主力净流入"的选股信号脚本，用 AkShare 双源取数、Pandas 清洗筛选、Rich 彩色终端渲染、openpyxl 持久化，辅以异常隔离、限速、进度条、汇总统计等工程化设计，把一次盘后选股任务做成了一条可读、可跑、可扩展的流水线。

四、脚本

#!/usr/bin/env python3
import akshare as ak
import pandas as pd
import time
from datetime import datetime
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn
from rich import box
from rich.text import Text

console = Console()

def format_amount(amount):
"""将金额格式化为易读的中文单位"""
if amount is None or pd.isna(amount):
return "N/A"
amount = float(amount)
if abs(amount) >= 1e8:
return f"{amount / 1e8:,.2f} 亿"
elif abs(amount) >= 1e4:
return f"{amount / 1e4:,.2f} 万"
else:
return f"{amount:,.0f}"

def find_new_lows_with_inflow():
# ———- 标题 ———-
console.print()
console.print(Panel.fit(
"[bold cyan]🔍 今日创历史新低 + 底部大资金抄底 筛选[/bold cyan]\\n"
f"[dim]运行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}[/dim]",
border_style="cyan",
box=box.ROUNDED
))

# ———- 1. 获取创新低股票 ———-
with console.status("[bold yellow]正在获取创历史新低股票列表…[/bold yellow]", spinner="dots"):
try:
low_df = ak.stock_rank_cxd_ths(symbol="历史新低")
except Exception as e:
console.print(f"[bold red]✗ 获取创新低数据失败:[/bold red] {e}")
return

if low_df is None or low_df.empty:
console.print("[bold yellow]⚠ 今日暂无创历史新低数据。[/bold yellow]")
return

stocks = low_df[['股票代码', '股票简称']].values.tolist()
total = len(stocks)

console.print(Panel(
f"[green]✓[/green] 共发现 [bold cyan]{total}[/bold cyan] 只创历史新低股票\\n"
f"[dim]开始逐一核查资金流向（主力净流入 > 0 视为大资金抄底）…[/dim]",
border_style="green",
box=box.ROUNDED
))

# ———- 2. 遍历核查资金流 ———-
results = []
errors = 0

with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(bar_width=40),
TextColumn("[cyan]{task.completed}/{task.total}[/cyan]"),
console=console,
transient=False,
) as progress:
task = progress.add_task("[yellow]核查资金流向…", total=total)

for code, name in stocks:
progress.update(task, description=f"[yellow]核查: {name} ({code})")

if code.startswith('6'):
market = 'sh'
elif code.startswith(('0', '3')):
market = 'sz'
elif code.startswith(('8', '4')):
market = 'bj'
else:
progress.advance(task)
continue

try:
flow_df = ak.stock_individual_fund_flow(stock=code, market=market)
if flow_df is None or flow_df.empty:
progress.advance(task)
continue

latest_flow = flow_df.iloc[-1]
main_net = latest_flow.get('主力净流入-净额', 0)
super_net = latest_flow.get('超大单净流入-净额', 0)
ratio = latest_flow.get('主力净流入-净占比', None)

if pd.notna(main_net) and main_net > 0:
results.append({
'代码': code,
'名称': name,
'收盘价': latest_flow.get('收盘价', None),
'涨跌幅': latest_flow.get('涨跌幅', None),
'主力净流入': main_net,
'超大单净流入': super_net,
'主力净占比': ratio,
'日期': latest_flow.get('日期', 'N/A'),
})

time.sleep(0.3)

except Exception:
errors += 1
finally:
progress.advance(task)

# ———- 3. 输出结果 ———-
console.print()

if not results:
console.print(Panel(
"[bold yellow]⚠ 未发现创历史新低且主力资金净流入的股票。[/bold yellow]\\n"
"[dim]提示：可能是今日市场无符合条件标的，或数据尚未更新。[/dim]",
border_style="yellow",
box=box.ROUNDED
))
return

# 按主力净流入降序排序
res_df = pd.DataFrame(results).sort_values('主力净流入', ascending=False).reset_index(drop=True)

# 构建美化表格
table = Table(
title=f"📊 筛选结果：{len(res_df)} 只标的符合「创新低 + 主力净流入」",
box=box.HEAVY_HEAD,
border_style="cyan",
header_style="bold white on dark_cyan",
title_style="bold cyan",
show_lines=True,
pad_edge=False,
)
table.add_column("序号", justify="center", style="dim", width=4)
table.add_column("代码", justify="center", style="bold white", width=8)
table.add_column("名称", justify="left", style="bold", width=10)
table.add_column("收盘价", justify="right", width=8)
table.add_column("涨跌幅", justify="right", width=8)
table.add_column("主力净流入", justify="right", width=12)
table.add_column("超大单净流入", justify="right", width=12)
table.add_column("主力净占比", justify="right", width=10)
table.add_column("日期", justify="center", style="dim", width=10)

for i, row in res_df.iterrows():
# 收盘价：红涨绿跌
price = row['收盘价']
pct = row['涨跌幅']
if pd.notna(pct):
pct_color = "red" if pct > 0 else ("green" if pct < 0 else "white")
pct_str = f"[{pct_color}]{pct:+.2f}%[/{pct_color}]"
else:
pct_str = "[dim]N/A[/dim]"

# 净流入：正红负绿
main_amt = row['主力净流入']
main_str = f"[bold red]{format_amount(main_amt)}[/bold red]" if main_amt > 0 \\
else f"[green]{format_amount(main_amt)}[/green]"

super_amt = row['超大单净流入']
if pd.notna(super_amt):
super_str = f"[red]{format_amount(super_amt)}[/red]" if super_amt > 0 \\
else f"[green]{format_amount(super_amt)}[/green]"
else:
super_str = "[dim]N/A[/dim]"

ratio = row['主力净占比']
ratio_str = f"{ratio:+.2f}%" if pd.notna(ratio) else "[dim]N/A[/dim]"

table.add_row(
str(i + 1),
row['代码'],
row['名称'],
f"{price:.2f}" if pd.notna(price) else "[dim]N/A[/dim]",
pct_str,
main_str,
super_str,
ratio_str,
str(row['日期']),
)

console.print(table)

# ———- 4. 汇总统计 ———-
total_inflow = res_df['主力净流入'].sum()
total_super = res_df['超大单净流入'].sum()
avg_ratio = res_df['主力净占比'].dropna().mean()

summary = Table(box=box.SIMPLE, show_header=False, padding=(0, 2))
summary.add_column(style="bold cyan", width=20)
summary.add_column(style="bold white")
summary.add_row("📌 符合条件标的数", f"{len(res_df)} 只")
summary.add_row("💰 主力净流入合计", f"[bold red]{format_amount(total_inflow)}[/bold red]")
summary.add_row("🏦 超大单净流入合计", f"[bold red]{format_amount(total_super)}[/bold red]")
summary.add_row("📈 主力净占比均值", f"{avg_ratio:+.2f}%" if pd.notna(avg_ratio) else "N/A")
summary.add_row("⚠️ 数据获取失败数", f"{errors} 只")

console.print(Panel(summary, title="[bold]📋 汇总统计[/bold]", border_style="magenta", box=box.ROUNDED))

# ———- 5. 导出 Excel ———-
try:
filename = f"new_low_inflow_{datetime.now().strftime('%Y%m%d_%H%M')}.xlsx"
res_df.to_excel(filename, index=False)
console.print(f"[green]✓ 结果已导出:[/green] [bold underline]{filename}[/bold underline]\\n")
except Exception as e:
console.print(f"[yellow]⚠ Excel 导出失败: {e}[/yellow]\\n")

if __name__ == "__main__":
find_new_lows_with_inflow()

五、效果截图