.\pandas-ta\pandas_ta\trend\xsignals.py
# -*- coding: utf-8 -*- # 从 numpy 中导入 nan 并重命名为 npNaN from numpy import nan as npNaN # 从 pandas 中导入 DataFrame from pandas import DataFrame # 从当前包中导入 tsignals 模块 from .tsignals import tsignals # 从 pandas_ta.utils._signals 中导入 cross_value 函数 from pandas_ta.utils._signals import cross_value # 从 pandas_ta.utils 中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义函数 xsignals，用于计算交叉信号 def xsignals(signal, xa, xb, above:bool=True, long:bool=True, asbool:bool=None, trend_reset:int=0, trade_offset:int=None, offset:int=None, **kwargs): """Indicator: Cross Signals""" # 验证参数 signal = verify_series(signal) offset = get_offset(offset) # 计算结果 if above: # 如果 above 为 True，计算 signal 与 xa 交叉的位置 entries = cross_value(signal, xa) # 计算 signal 与 xb 交叉的位置，注意指定 above=False exits = -cross_value(signal, xb, above=False) else: # 如果 above 为 False，计算 signal 与 xa 交叉的位置，注意指定 above=False entries = cross_value(signal, xa, above=False) # 计算 signal 与 xb 交叉的位置 exits = -cross_value(signal, xb) # 计算交叉信号 trades = entries + exits # 修改交叉信号以填充趋势间的间隙 trades.replace({0: npNaN}, inplace=True) trades.interpolate(method="pad", inplace=True) trades.fillna(0, inplace=True) # 将交叉信号转换为趋势 trends = (trades > 0).astype(int) if not long: trends = 1 - trends # 构建传递给 tsignals 函数的关键字参数字典 tskwargs = { "asbool":asbool, "trade_offset":trade_offset, "trend_reset":trend_reset, "offset":offset } # 调用 tsignals 函数计算趋势信号 df = tsignals(trends, **tskwargs) # 处理偏移，由 tsignals 函数处理 DataFrame({ f"XS_LONG": df.TS_Trends, f"XS_SHORT": 1 - df.TS_Trends }) # 处理填充 if "fillna" in kwargs: df.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: df.fillna(method=kwargs["fill_method"], inplace=True) # 设定名称和类别 df.name = f"XS" df.category = "trend" return df # 设定函数文档字符串 xsignals.__doc__ = \ """Cross Signals (XSIGNALS) Cross Signals returns Trend Signal (TSIGNALS) results for Signal Crossings. This is useful for indicators like RSI, ZSCORE, et al where one wants trade Entries and Exits (and Trends). Cross Signals has two kinds of modes: above and long. The first mode 'above', default True, xsignals determines if the signal first crosses above 'xa' and then below 'xb'. If 'above' is False, xsignals determines if the signal first crosses below 'xa' and then above 'xb'. The second mode 'long', default True, passes the long trend result into tsignals so it can determine the appropriate Entries and Exits. When 'long' is False, it does the same but for the short side. Example: # These are two different outcomes and depends on the indicator and it's # characteristics. Please check BOTH outcomes BEFORE making an Issue. rsi = df.ta.rsi() # Returns tsignal DataFrame when RSI crosses above 20 and then below 80 ta.xsignals(rsi, 20, 80, above=True) # Returns tsignal DataFrame when RSI crosses below 20 and then above 80 ta.xsignals(rsi, 20, 80, above=False) Source: Kevin Johnson Calculation: Default Inputs: asbool=False, trend_reset=0, trade_offset=0, drift=1 trades = trends.diff().shift(trade_offset).fillna(0).astype(int) entries = (trades > 0).astype(int) exits = (trades < 0).abs().astype(int) Args: """ # 定义一个布尔值，表示信号是在'xa'之上首次穿越，然后再穿越'xb'，还是在'xa'之下首次穿越，然后再穿越'xb' above (bool): When the signal crosses above 'xa' first and then 'xb'. When False, then when the signal crosses below 'xa' first and then 'xb'. Default: True # 将长期趋势传递给tsignals的趋势参数。当为False时，将短期趋势传递给tsignals的趋势参数 long (bool): Passes the long trend into tsignals' trend argument. When False, it passes the short trend into tsignals trend argument. Default: True # 差异期。默认值为1 drift (int): The difference period. Default: 1 # 结果的偏移量。默认值为0 offset (int): How many periods to offset the result. Default: 0 # TSIGNAL传递参数 # 如果为True，则将Trends、Entries和Exits列转换为布尔值。当为布尔值时，也可用于使用vectorbt的Portfolio.from_signal(close, entries, exits)进行回测 asbool (bool): If True, it converts the Trends, Entries and Exits columns to booleans. When boolean, it is also useful for backtesting with vectorbt's Portfolio.from_signal(close, entries, exits) Default: False # 用于识别趋势是否结束的值。默认值为0 trend_reset (value): Value used to identify if a trend has ended. Default: 0 # 用于移动交易进出的值。使用1进行回测，使用0进行实时交易。默认值为0 trade_offset (value): Value used shift the trade entries/exits Use 1 for backtesting and 0 for live. Default: 0 # 函数参数说明，使用关键字参数传递给函数的参数列表 Kwargs: # fillna参数，用于填充缺失值的值，采用pd.DataFrame.fillna(value)方式 fillna (value, optional): pd.DataFrame.fillna(value) # fill_method参数，填充缺失值的方法类型 fill_method (value, optional): Type of fill method # 返回值说明，返回一个pd.DataFrame对象，其包含以下列： Returns: # Trends列，趋势（有趋势: 1，无趋势: 0） Trends (trend: 1, no trend: 0), # Trades列，交易（进入: 1，退出: -1，其他: 0） Trades (Enter: 1, Exit: -1, Otherwise: 0), # Entries列，入口（入口: 1，无: 0） Entries (entry: 1, nothing: 0), # Exits列，出口（出口: 1，无: 0） Exits (exit: 1, nothing: 0)
.\pandas-ta\pandas_ta\trend\__init__.py
# -*- coding: utf-8 -*- # 指定文件编码为 UTF-8，确保正确处理中文字符 # 导入各个指标模块 from .adx import adx # 导入 adx 指标模块 from .amat import amat # 导入 amat 指标模块 from .aroon import aroon # 导入 aroon 指标模块 from .chop import chop # 导入 chop 指标模块 from .cksp import cksp # 导入 cksp 指标模块 from .decay import decay # 导入 decay 指标模块 from .decreasing import decreasing # 导入 decreasing 指标模块 from .dpo import dpo # 导入 dpo 指标模块 from .increasing import increasing # 导入 increasing 指标模块 from .long_run import long_run # 导入 long_run 指标模块 from .psar import psar # 导入 psar 指标模块 from .qstick import qstick # 导入 qstick 指标模块 from .short_run import short_run # 导入 short_run 指标模块 from .tsignals import tsignals # 导入 tsignals 指标模块 from .ttm_trend import ttm_trend # 导入 ttm_trend 指标模块 from .vhf import vhf # 导入 vhf 指标模块 from .vortex import vortex # 导入 vortex 指标模块 from .xsignals import xsignals # 导入 xsignals 指标模块
.\pandas-ta\pandas_ta\utils\data\alphavantage.py
# -*- coding: utf-8 -*- # 导入 DataFrame 类 from pandas import DataFrame # 导入 Imports 对象，RATE 对象，version 对象 from pandas_ta import Imports, RATE, version # 定义 av 函数，获取 alphaVantage 数据 def av(ticker: str, **kwargs): # 打印关键字参数 kwargs print(f"[!] kwargs: {kwargs}") # 从 kwargs 中弹出 verbose 参数，默认为 False verbose = kwargs.pop("verbose", False) # 从 kwargs 中弹出 kind 参数，默认为 "history" kind = kwargs.pop("kind", "history") # 将 kind 转换为小写 kind = kind.lower() # 从 kwargs 中弹出 interval 参数，默认为 "D" interval = kwargs.pop("interval", "D") # 从 kwargs 中弹出 show 参数，默认为 None show = kwargs.pop("show", None) # 从 kwargs 中弹出 last 参数，但是没有使用到 # 如果 ticker 不为空且是字符串类型，则将其转换为大写，否则为 None ticker = ticker.upper() if ticker is not None and isinstance(ticker, str) else None # 如果 alphaVantage-api 可用且 ticker 不为空 if Imports["alphaVantage-api"] and ticker is not None: # 导入 alphaVantageAPI 模块并重命名为 AV import alphaVantageAPI as AV # 定义 AVC 字典，包含 API 密钥和其他参数 AVC = {"api_key": "YOUR API KEY", "clean": True, "export": False, "output_size": "full", "premium": False} # 从 kwargs 中获取 av_kwargs 参数，如果不存在则使用 AVC _config = kwargs.pop("av_kwargs", AVC) # 创建 AlphaVantage 对象 av av = AV.AlphaVantage(**_config) # 从 kwargs 中获取 period 参数，默认为 av.output_size period = kwargs.pop("period", av.output_size) # 定义 _all 列表和 div 变量 _all, div = ["all"], "=" * 53 # Max div width is 80 # 如果 kind 在 _all 列表中或者 verbose 为真，则执行下面的代码 if kind in _all or verbose: pass # 如果 kind 在 _all 列表或者 ["history", "h"] 列表中 if kind in _all + ["history", "h"]: # 如果 verbose 为真 if verbose: # 打印信息，显示 Pandas TA 版本和 alphaVantage-api print("\n==== Chart History " + div + f"\n[*] Pandas TA v{version} & alphaVantage-api") # 打印下载信息，显示下载的股票信息和时间间隔 print(f"[+] Downloading {ticker}[{interval}:{period}] from {av.API_NAME} (https://www.alphavantage.co/)") # 获取股票数据并保存到 df 变量中 df = av.data(ticker, interval) # 设置 DataFrame 的名称为 ticker df.name = ticker # 如果 show 不为空且是正整数且大于 0 if show is not None and isinstance(show, int) and show > 0: # 打印 DataFrame 最后几行数据 print(f"\n{df.name}\n{df.tail(show)}\n") # 返回 DataFrame 对象 return df # 如果上述条件都不满足，则返回一个空的 DataFrame 对象 return DataFrame()
# `.\pandas-ta\pandas_ta\utils\data\yahoofinance.py` ```py # -*- coding: utf-8 -*- # 导入 DataFrame 类 from pandas import DataFrame # 导入 Imports、RATE、version 变量 from pandas_ta import Imports, RATE, version # 导入 _camelCase2Title 函数和 ytd 函数 from .._core import _camelCase2Title from .._time import ytd # 定义函数 yf，用于包装 yfinance def yf(ticker: str, **kwargs): """yf - yfinance wrapper It retrieves market data (ohlcv) from Yahoo Finance using yfinance. To install yfinance. (pip install yfinance) This method can also pull additional data using the 'kind' kwarg. By default kind=None and retrieves Historical Chart Data. Other options of 'kind' include: * All: "all" - Prints everything below but only returns Chart History to Pandas TA * Company Information: "info" * Institutional Holders: "institutional_holders" or "ih" * Major Holders: "major_holders" or "mh" * Mutual Fund Holders: "mutualfund_holders" or "mfh" * Recommendations (YTD): "recommendations" or "rec" * Earnings Calendar: "calendar" or "cal" * Earnings: "earnings" or "earn" * Sustainability/ESG Scores: "sustainability", "sus" or "esg" * Financials: "financials" or "fin" - Returns in order: Income Statement, Balance Sheet and Cash Flow * Option Chain: "option_chain" or "oc" - Uses the nearest expiration date by default - Change the expiration date using kwarg "exp" - Show ITM options, set kwarg "itm" to True. Or OTM options, set kwarg "itm" to False. * Chart History: - The only data returned to Pandas TA. Args: ticker (str): Any string for a ticker you would use with yfinance. Default: "SPY" Kwargs: calls (bool): When True, prints only Option Calls for the Option Chain. Default: None desc (bool): Will print Company Description when printing Company Information. Default: False exp (str): Used to print other Option Chains for the given Expiration Date. Default: Nearest Expiration Date for the Option Chains interval (str): A yfinance argument. Default: "1d" itm (bool): When printing Option Chains, shows ITM Options when True. When False, it shows OTM Options: Default: None kind (str): Options see above. Default: None period (str): A yfinance argument. Default: "max" proxy (dict): Proxy for yfinance to use. Default: {} puts (bool): When True, prints only Option Puts for the Option Chain. Default: None show (int > 0): How many last rows of Chart History to show. Default: None snd (int): How many recent Splits and Dividends to show in Company Information. Default: 5 verbose (bool): Prints Company Information "info" and a Chart History header to the screen. Default: False Returns: Exits if the DataFrame is empty or None Otherwise it returns a DataFrame of the Chart History """ # 从 kwargs 中获取 verbose 参数，默认为 False verbose = kwargs.pop("verbose", False) # 如果 ticker 不为空且为字符串类型且长度大于0，则将 ticker 转换为大写 if ticker is not None and isinstance(ticker, str) and len(ticker): ticker = ticker.upper() else: # 如果 ticker 为空或不是字符串类型或长度为0，则将 ticker 设置为 "SPY" ticker = "SPY" # 从 kwargs 中弹出 "kind" 键对应的值，如果不存在则为 None kind = kwargs.pop("kind", None) # 如果 kind 不为空且为字符串类型且长度大于0，则将 kind 转换为小写 if kind is not None and isinstance(kind, str) and len(kind): kind = kind.lower() # 从 kwargs 中弹出 "period" 键对应的值，如果不存在则为 "max" period = kwargs.pop("period", "max") # 从 kwargs 中弹出 "interval" 键对应的值，如果不存在则为 "1d" interval = kwargs.pop("interval", "1d") # 从 kwargs 中弹出 "proxy" 键对应的值，如果不存在则为一个空字典 proxy = kwargs.pop("proxy", {}) # 从 kwargs 中弹出 "show" 键对应的值，如果不存在则为 None show = kwargs.pop("show", None) # 如果 Imports 中没有 yfinance 模块，则打印提示信息并返回 if not Imports["yfinance"]: print(f"[X] Please install yfinance to use this method. (pip install yfinance)") return else: # 如果有 yfinance 模块，则返回一个空的 DataFrame 对象 return DataFrame()
.\pandas-ta\pandas_ta\utils\data\__init__.py
# 设置文件编码为 UTF-8，以支持包含非 ASCII 字符的内容 # 导入自定义模块中的 alphavantage 和 yahoofinance 子模块 from .alphavantage import av from .yahoofinance import yf
.\pandas-ta\pandas_ta\utils\_candles.py
# -*- coding: utf-8 -*- # 导入 Series 类 from pandas import Series # 导入 non_zero_range 函数 from ._core import non_zero_range # 计算蜡烛图的颜色 def candle_color(open_: Series, close: Series) -> Series: # 复制收盘价 Series，并将其类型转换为整数 color = close.copy().astype(int) # 当收盘价大于等于开盘价时，将颜色设置为1 color[close >= open_] = 1 # 当收盘价小于开盘价时，将颜色设置为-1 color[close < open_] = -1 # 返回颜色 Series return color # 计算最高价和最低价的范围 def high_low_range(high: Series, low: Series) -> Series: # 调用 non_zero_range 函数计算高低价的范围 return non_zero_range(high, low) # 计算实体部分（实体部分指收盘价与开盘价之间的绝对值） def real_body(open_: Series, close: Series) -> Series: # 调用 non_zero_range 函数计算实体部分 return non_zero_range(close, open_)
PandasTA 源码解析（十四）（2）https://developer.aliyun.com/article/1506234
