当你点一杯"大杯冰燕麦拿铁,不加糖"时,咖啡师并不会把你的原话记下来,而是在杯子上勾选四个选项。本实战指南对交易 API 做的是同样的事情:一句话进去,出来的是一个函数名及其参数——已被求值的枚举,每个都带一个置信度。
"plot rolling correlation between nvda and spy for the past month"
rolling_correlation(symbol='NVDA', benchmark='SPY', window='1mo') confidence 0.91
"compare nvda amd and msft over the past three months"
compare_returns(symbols=['NVDA', 'AMD', 'MSFT'], window='3mo') confidence 0.94
"show me apple daily with volume"
plot_price(symbol='AAPL', resolution='1d', include_volume=True) confidence 0.75
"what tickers do you have"
list_symbols() confidence 1.00
这些调用会交给交易助手中的十个普通函数。这些函数的参数取值来自固定列表,因此它们本身就是 Literal:
def plot_price(
symbol: Literal["SPY", "NVDA", "AMD", "AAPL", "MSFT", "TSLA"],
style: Literal["line", "candles"] = "line",
resolution: Literal["1m", "5m", "15m", "1h", "1d"] = "15m",
window: Literal["1d", "1w", "1mo", "3mo"] = "1w",
include_volume: bool = False,
moving_average: Literal["9", "20", "50"] | None = None,
log_scale: bool = False,
): ...
取值来自固定列表的参数就是一个闭集(closed set)。当它从列表中取一个值时,会得到一个恰好覆盖这些取值的 Choice 问题,因此传到函数手里的总是函数能接受的值。函数本身原封不动;你要添加的只是一份用平实语言说明每个参数含义的规格说明(spec)。到最后,你会得到一个可以指向你自己函数的 Dispatcher。
准备工作
pip install ipython polars matplotlib numpy "typesafe-sdk>=0.5.7" cooksafe --extra-index-url https://pypi.typesafe.ai/
设置 TYPESAFE_API_KEY。本文件旁边有两个模块。trader.py 包含那十个函数,以及一个从缓存读取答案的 TypeSafe 客户端,因此重新渲染会重放下面这些数字,而不会调用 API。dispatch.py 包含读取签名和规格说明并发起调用的代码。
import json
from pathlib import Path
from cooksafe import make_playground_link
from dispatch import ROUTE, Dispatcher, closed_sets
from IPython.display import Markdown, display
from trader import TOOLS, client, load
TYPESAFE_MODEL = "jev-1.12"
print(f"{len(TOOLS)} functions over {load().height:,} one-minute bars")
10 functions over 156,780 one-minute bars
从签名中找出闭集
类型注解已经说明了哪些参数来自固定列表,以及每个列表里有什么。closed_sets 读取一个签名,并把这些参数归入三种形态:choice(一个 Literal,即从列表中取一个值)、set(一个 list[Literal[...]],即取任意多个值)、或 flag(一个 bool,即开或关)。十个函数全部定义在 trader.py 中。
for name, fn in TOOLS.items():
shapes = closed_sets(fn)
print(
f" {name:<20}{len(shapes)} "
+ ", ".join(f"{a}:{s}" for a, (s, _) in shapes.items())
)
print(
f"\n{sum(len(closed_sets(fn)) for fn in TOOLS.values())} fillable arguments in total"
)
list_symbols 0
market_summary 1 window:choice
plot_price 7 symbol:choice, style:choice, resolution:choice, window:choice, include_volume:flag, moving_average:choice, log_scale:flag
intraday_pattern 3 symbol:choice, window:choice, metric:choice
compare_returns 3 symbols:set, window:choice, normalize:flag
rolling_correlation 4 symbol:choice, benchmark:choice, window:choice, resolution:choice
summary_stats 2 symbol:choice, window:choice
volatility 3 symbol:choice, window:choice, annualized:flag
top_movers 2 window:choice, direction:choice
drawdown 3 symbol:choice, window:choice, plot:flag
28 fillable arguments in total
top_movers 展示了哪些会被排除在外。它的三个参数中有两个是闭集,第三个 limit 是 int,因此它永远不会得到问题,并保持默认值 3。自由文本、数字和日期的处理方式相同:没有问题,函数的默认值照常生效。
编写规格说明
Literal 给了你字符串 "1mo" 和 "3mo",但它并没有说明用户输入 "this quarter" 指的是后者。规格说明负责说清这一点。它为每个参数保存一个问题,为每个选项保存一行说明,为每个函数保存一段描述,另外还有一个用来在函数之间做选择的问题。它存放在 spec.json 中,LLM 可以根据签名替你把它写出来。
SPEC = json.loads(Path("spec.json").read_text())
for argument in ("style", "moving_average"):
print(
json.dumps(
{argument: SPEC["functions"]["plot_price"]["arguments"][argument]}, indent=2
)
)
{
"style": {
"question": "Does the user want a plain line or candles?",
"stated": "Does the user say how the chart should be drawn, such as a line, candles, or OHLC bars?",
"options": {
"line": "a simple line through the closing prices",
"candles": "a candlestick or OHLC chart, showing each bar's open, high, low and close"
}
}
}
{
"moving_average": {
"question": "How many bars should the moving average cover - nine, twenty, or fifty?",
"stated": "Does the user ask for a moving average or a smoothed line over the candles?",
"options": {
"9": "a nine-bar moving average, a fast one",
"20": "a twenty-bar moving average",
"50": "a fifty-bar moving average, a slow one"
}
}
}
选项的键就是函数接受的字符串,因此之后不需要把某个标签再映射回参数。stated 让一个参数变成可选的。它是另一个是/否问题,询问命令究竟有没有提到这个参数。当答案为否时,调用会省略该参数,函数自身的默认值随之生效。
set 型参数的问题按成员各生成一次,{} 代表成员名。"Does the user want {} in the comparison?" 会变成每个股票代码各一个问题。
每个问题都应围绕概念来写,而不是围绕用户可能选用的字眼,因为匹配是基于语义的:"is amd tracking nvidia lately" 能命中 rolling_correlation,尽管 tracking 和 lately 都没有出现在 spec.json 的任何地方。不要用参数名来命名问题——"Which resolution?" 没有给命令提供任何可匹配的内容。
把规格说明变成问题
Dispatcher 会根据规格说明一次性构建好问题。之后每条命令都是一次请求,携带函数选择和所有函数的参数问题,而分发器只读取所选函数的答案。
assistant = Dispatcher(SPEC, TOOLS, client)
print(f"{len(assistant.questions)} questions per command, among them:")
for qid in (
"__tool__",
"plot_price.style",
"plot_price.style?",
"compare_returns.symbols.NVDA",
):
question = assistant.questions[qid]
print(f" {qid:<30}{question['type']:<8}{str(question['instructions'])[:64]}")
54 questions per command, among them:
__tool__ choice What is the user asking the trading assistant to do?
plot_price.style choice Does the user want a plain line or candles?
plot_price.style? noul Does the user say how the chart should be drawn, such as a line,
compare_returns.symbols.NVDA noul Does the user want NVDA in the comparison?
运行十四条命令
每个请求占一行,其 confidence 是这次调用背后置信度最低的那个判断。
COMMANDS = [
"show nvda 1h",
"plot rolling correlation between nvda and spy for the past month",
"when during the day does nvda trade the most",
"what moved today",
"what tickers do you have",
"how did the market do this week",
"candles for tesla with a 20 period moving average",
"compare nvda amd and msft over the past three months",
"how volatile is tsla",
"biggest losers today",
"worst drawdown for nvda this quarter, and chart it please",
"spy stats for the last month",
"show me apple daily with volume",
"is amd tracking nvidia lately",
]
CALLS = {command: assistant(command) for command in COMMANDS}
for command, call in CALLS.items():
print(f' "{command}"')
print(
f" {str(call):<66}confidence {call.confidence:.2f}"
f" tool {call.tool.probability:.2f}"
)
"show nvda 1h"
plot_price(symbol='NVDA', resolution='1h') confidence 0.78 tool 1.00
"plot rolling correlation between nvda and spy for the past month"
rolling_correlation(symbol='NVDA', benchmark='SPY', window='1mo') confidence 0.91 tool 1.00
"when during the day does nvda trade the most"
intraday_pattern(symbol='NVDA') confidence 0.53 tool 1.00
"what moved today"
top_movers(window='1d', direction='gainers') confidence 0.90 tool 0.90
"what tickers do you have"
list_symbols() confidence 1.00 tool 1.00
"how did the market do this week"
market_summary(window='1w') confidence 0.96 tool 0.99
"candles for tesla with a 20 period moving average"
plot_price(symbol='TSLA', style='candles', moving_average='20') confidence 0.69 tool 0.97
"compare nvda amd and msft over the past three months"
compare_returns(symbols=['NVDA', 'AMD', 'MSFT'], window='3mo') confidence 0.94 tool 1.00
"how volatile is tsla"
volatility(symbol='TSLA') confidence 0.96 tool 1.00
"biggest losers today"
top_movers(window='1d', direction='losers') confidence 0.98 tool 0.98
"worst drawdown for nvda this quarter, and chart it please"
drawdown(symbol='NVDA', window='3mo', plot=True) confidence 0.84 tool 0.84
"spy stats for the last month"
summary_stats(symbol='SPY', window='1mo') confidence 0.88 tool 0.88
"show me apple daily with volume"
plot_price(symbol='AAPL', resolution='1d', include_volume=True) confidence 0.75 tool 0.85
"is amd tracking nvidia lately"
rolling_correlation(symbol='AMD', benchmark='NVDA') confidence 0.82 tool 0.82
两条长命令都按要求得到了结果。"plot rolling correlation between nvda and spy for the past month" 从一句话中填好了四个参数。其中 symbol 和 benchmark 取自同样的六个股票代码,而每个代码都落到了正确的参数上,因为问题把角色讲得明明白白:被度量的那一个,名字在前,对上第二个被提到的,也就是用作尺子的那一个。"compare nvda amd and msft over the past three months" 把三个代码放进了集合,把另外三个留在了外面。
运行其中三条:
for command in (
"plot rolling correlation between nvda and spy for the past month",
"compare nvda amd and msft over the past three months",
"when during the day does nvda trade the most",
):
print(f'"{command}" -> {CALLS[command]}')
display(CALLS[command].run())
"plot rolling correlation between nvda and spy for the past month" -> rolling_correlation(symbol='NVDA', benchmark='SPY', window='1mo')
"compare nvda amd and msft over the past three months" -> compare_returns(symbols=['NVDA', 'AMD', 'MSFT'], window='3mo')
"when during the day does nvda trade the most" -> intraday_pattern(symbol='NVDA')
再看看以文本作答的那几条:
for command in ("how did the market do this week", "biggest losers today"):
print(f'"{command}" -> {CALLS[command]}')
print(CALLS[command].run(), "\n")
"how did the market do this week" -> market_summary(window='1w')
the board over 1w
NVDA 254.12 9.62% 389,465,563
AMD 184.20 1.51% 182,740,497
AAPL 258.71 0.97% 223,818,998
SPY 664.86 0.40% 138,617,365
MSFT 451.35 0.26% 113,427,173
TSLA 320.22 -0.97% 266,317,023
"biggest losers today" -> top_movers(window='1d', direction='losers')
top 3 losers over 1d
AMD -0.57% -> 184.20
MSFT 0.67% -> 451.35
AAPL 1.40% -> 258.71
读懂置信度
confidence 报告的是调用中置信度最低的那个判断,而不是所有判断的乘积,因为只要一个参数错了就足以毁掉结果。乘积回答的是另一个问题("是不是每一部分都对"),而且无论单个判断是否动摇,它都会随着函数参数的增多而下降。
这个数字从何而来,逐参数来看:
call = CALLS["is amd tracking nvidia lately"]
print(f'"is amd tracking nvidia lately" -> {call} confidence {call.confidence:.2f}')
for name, argument in call.arguments.items():
top = sorted(argument.distribution.items(), key=lambda kv: -kv[1])[:3]
shown = "omitted, default stands" if argument.omitted else repr(argument.value)
print(
f" {name:<12}{shown:<26}p {argument.probability:.2f} "
+ " ".join(f"{k} {v:.2f}" for k, v in top)
)
print(f" weakest argument: {call.weakest().name}")
"is amd tracking nvidia lately" -> rolling_correlation(symbol='AMD', benchmark='NVDA') confidence 0.82
symbol 'AMD' p 0.87 AMD 0.87 NVDA 0.13 AAPL 0.00
benchmark 'NVDA' p 0.78 NVDA 0.92 AMD 0.08 AAPL 0.00
window omitted, default stands p 0.96
resolution omitted, default stands p 0.99
weakest argument: benchmark
这里 window 和 resolution 都被省略了,因为 "lately" 没有说明回看多远、用什么周期的 K 线,于是 rolling_correlation 按它自己的默认值运行:一个月、小时线。这正是 stated 问题的用途。没有它,choice 问题将不得不指定某个时间窗口,而且它会自信满满地指定一个。
在 Playground 中打开
下面的链接包含一条命令和它所选函数的问题:对十个函数描述的 choice,以及 rolling_correlation 的四个参数。在那里编辑命令,参数也会随之变化。
COMMAND = "plot rolling correlation between nvda and spy for the past month"
picked = CALLS[COMMAND]
playground_link = make_playground_link(
COMMAND,
{ROUTE: assistant.questions[ROUTE]}
| {q: v for q, v in assistant.questions.items() if q.startswith(f"{picked.name}.")},
models=[TYPESAFE_MODEL],
)
display(
Markdown(
f"🔗 [Open the command and its questions in the TypeSafe playground]({playground_link})"
)
)